Added Workflows (#359)

* Implemented Workflows for OpenDC

Author: DanteNiewenhuis

opendc-compute/opendc-compute-api/src/main/kotlin/org/opendc/compute/api/Flavor.kt

@@ -41,4 +41,19 @@ public interface Flavor : Resource {
      * The amount of gpu cores available to the task.
      */
     public val gpuCoreCount: Int
+
+    /**
+     * Set of Tasks that need to be finished before this can startAdd commentMore actions
+     */
+    public val dependencies: Set<String>
+
+    /**
+     * Set of Tasks that need to be finished before this can startAdd commentMore actions
+     */
+    public val parents: Set<String>
+
+    /**
+     * Set of Tasks that need to be finished before this can startAdd commentMore actions
+     */
+    public val children: Set<String>
 }

opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java

@@ -119,16 +119,19 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
      */
     private final Deque<SchedulingRequest> taskQueue = new ArrayDeque<>();
 
+    private final List<SchedulingRequest> blockedTasks = new ArrayList<>();
+
     /**
      * The active tasks in the system.
      */
     private final Map<ServiceTask, SimHost> activeTasks = new HashMap<>();
 
     /**
      * The active tasks in the system.
-     * TODO: this is not doing anything, maybe delete it?
      */
-    private final Map<ServiceTask, SimHost> completedTasks = new HashMap<>();
+    private final List<String> completedTasks = new ArrayList<>();
+
+    private final List<String> terminatedTasks = new ArrayList<>();
 
     /**
      * The registered flavors for this compute service.
@@ -209,9 +212,11 @@ public void onStateChanged(@NotNull SimHost host, @NotNull ServiceTask task, @No
 
                 if (newState == TaskState.COMPLETED) {
                     tasksCompleted++;
+                    addCompletedTask(task);
                 }
                 if (newState == TaskState.TERMINATED) {
                     tasksTerminated++;
+                    addTerminatedTask(task);
                 }
 
                 if (task.getState() == TaskState.COMPLETED || task.getState() == TaskState.TERMINATED) {
@@ -430,17 +435,83 @@ SchedulingRequest schedule(ServiceTask task, boolean atFront) {
         long now = clock.millis();
         SchedulingRequest request = new SchedulingRequest(task, now);
 
-        if (atFront) {
-            taskQueue.addFirst(request);
-        } else {
-            taskQueue.add(request);
+        ServiceFlavor flavor = task.getFlavor();
+        for (String taskName : this.terminatedTasks) {
+            if (flavor.isInDependencies(taskName)) {
+                // Terminate task
+                task.setState(TaskState.TERMINATED);
+            }
         }
+
+        // Remove all completed tasks from the pending dependencies
+        flavor.updatePendingDependencies(this.completedTasks);
+
+        // If there are still pending dependencies, we cannot schedule the task yet
+        Set<String> pendingDependencies = flavor.getDependencies();
+        if (!pendingDependencies.isEmpty()) {
+            // If the task has pending dependencies, we cannot schedule it yet
+            LOGGER.debug("Task {} has pending dependencies: {}", task.getUid(), pendingDependencies);
+            blockedTasks.add(request);
+            return null;
+        }
+
+        // Add the request at the front or the back of the queue
+        if (atFront) taskQueue.addFirst(request);
+        else taskQueue.add(request);
+
         tasksPending++;
 
         requestSchedulingCycle();
         return request;
     }
 
+    void addCompletedTask(ServiceTask task) {
+        String taskName = task.getName();
+
+        if (!this.completedTasks.contains(taskName)) {
+            this.completedTasks.add(taskName);
+        }
+
+        List<SchedulingRequest> requestsToRemove = new ArrayList<>();
+
+        for (SchedulingRequest request : blockedTasks) {
+            request.getTask().getFlavor().updatePendingDependencies(taskName);
+
+            Set<String> pendingDependencies = request.getTask().getFlavor().getDependencies();
+
+            if (pendingDependencies.isEmpty()) {
+                requestsToRemove.add(request);
+                taskQueue.add(request);
+                tasksPending++;
+            }
+        }
+
+        for (SchedulingRequest request : requestsToRemove) {
+            blockedTasks.remove(request);
+        }
+    }
+
+    void addTerminatedTask(ServiceTask task) {
+        String taskName = task.getName();
+
+        List<SchedulingRequest> requestsToRemove = new ArrayList<>();
+
+        if (!this.terminatedTasks.contains(taskName)) {
+            this.terminatedTasks.add(taskName);
+        }
+
+        for (SchedulingRequest request : blockedTasks) {
+            if (request.getTask().getFlavor().isInDependencies(taskName)) {
+                requestsToRemove.add(request);
+                request.getTask().setState(TaskState.TERMINATED);
+            }
+        }
+
+        for (SchedulingRequest request : requestsToRemove) {
+            blockedTasks.remove(request);
+        }
+    }
+
     void delete(ServiceFlavor flavor) {
         flavorById.remove(flavor.getUid());
         flavors.remove(flavor);
@@ -612,12 +683,19 @@ public Flavor findFlavor(@NotNull UUID id) {
 
         @NotNull
         public ServiceFlavor newFlavor(
-                @NotNull String name, int cpuCount, long memorySize, int gpuCoreCount, @NotNull Map<String, ?> meta) {
+                @NotNull String name,
+                int cpuCount,
+                long memorySize,
+                int gpuCoreCount,
+                @NotNull Set<String> parents,
+                @NotNull Set<String> children,
+                @NotNull Map<String, ?> meta) {
             checkOpen();
 
             final ComputeService service = this.service;
             UUID uid = new UUID(service.clock.millis(), service.random.nextLong());
-            ServiceFlavor flavor = new ServiceFlavor(service, uid, name, cpuCount, memorySize, gpuCoreCount, meta);
+            ServiceFlavor flavor =
+                    new ServiceFlavor(service, uid, name, cpuCount, memorySize, gpuCoreCount, parents, children, meta);
 
             //            service.flavorById.put(uid, flavor);
             //            service.flavors.add(flavor);

opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceFlavor.java

@@ -23,8 +23,11 @@
 package org.opendc.compute.simulator.service;
 
 import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.UUID;
 import org.jetbrains.annotations.NotNull;
 import org.opendc.compute.api.Flavor;
@@ -39,6 +42,9 @@ public final class ServiceFlavor implements Flavor {
     private final int cpuCoreCount;
     private final long memorySize;
     private final int gpuCoreCount;
+    private final Set<String> parents;
+    private final Set<String> children;
+    private final Set<String> dependencies;
     private final Map<String, ?> meta;
 
     ServiceFlavor(
@@ -48,13 +54,18 @@ public final class ServiceFlavor implements Flavor {
             int cpuCoreCount,
             long memorySize,
             int gpuCoreCount,
+            Set<String> parents,
+            Set<String> children,
             Map<String, ?> meta) {
         this.service = service;
         this.uid = uid;
         this.name = name;
         this.cpuCoreCount = cpuCoreCount;
         this.memorySize = memorySize;
         this.gpuCoreCount = gpuCoreCount;
+        this.parents = parents;
+        this.dependencies = new HashSet<>(parents);
+        this.children = children;
         this.meta = meta;
     }
 
@@ -118,4 +129,33 @@ public int hashCode() {
     public String toString() {
         return "Flavor[uid=" + uid + ",name=" + name + "]";
     }
+
+    @Override
+    public @NotNull Set<String> getDependencies() {
+        return dependencies;
+    }
+
+    public void updatePendingDependencies(List<String> completedTasks) {
+        for (String task : completedTasks) {
+            this.updatePendingDependencies(task);
+        }
+    }
+
+    public void updatePendingDependencies(String completedTask) {
+        this.dependencies.remove(completedTask);
+    }
+
+    public boolean isInDependencies(String task) {
+        return this.dependencies.contains(task);
+    }
+
+    @Override
+    public @NotNull Set<@NotNull String> getParents() {
+        return parents;
+    }
+
+    @Override
+    public @NotNull Set<@NotNull String> getChildren() {
+        return children;
+    }
 }

opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt

@@ -29,16 +29,17 @@ import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy
 import org.opendc.trace.Trace
 import org.opendc.trace.conv.TABLE_RESOURCES
 import org.opendc.trace.conv.TABLE_RESOURCE_STATES
+import org.opendc.trace.conv.resourceChildren
 import org.opendc.trace.conv.resourceCpuCapacity
 import org.opendc.trace.conv.resourceCpuCount
 import org.opendc.trace.conv.resourceDeadline
 import org.opendc.trace.conv.resourceDuration
 import org.opendc.trace.conv.resourceGpuCapacity
 import org.opendc.trace.conv.resourceGpuCount
-import org.opendc.trace.conv.resourceGpuMemCapacity
 import org.opendc.trace.conv.resourceID
 import org.opendc.trace.conv.resourceMemCapacity
 import org.opendc.trace.conv.resourceNature
+import org.opendc.trace.conv.resourceParents
 import org.opendc.trace.conv.resourceStateCpuUsage
 import org.opendc.trace.conv.resourceStateDuration
 import org.opendc.trace.conv.resourceStateGpuUsage
@@ -136,7 +137,8 @@ public class ComputeWorkloadLoader(
         val memCol = reader.resolve(resourceMemCapacity)
         val gpuCapacityCol = reader.resolve(resourceGpuCapacity) // Assuming GPU capacity is also present
         val gpuCoreCountCol = reader.resolve(resourceGpuCount) // Assuming GPU cores are also present
-        val gpuMemoryCol = reader.resolve(resourceGpuMemCapacity) // Assuming GPU memory is also present
+        val parentsCol = reader.resolve(resourceParents)
+        val childrenCol = reader.resolve(resourceChildren)
         val natureCol = reader.resolve(resourceNature)
         val deadlineCol = reader.resolve(resourceDeadline)
 
@@ -166,6 +168,10 @@ public class ComputeWorkloadLoader(
                     }
                 val gpuCoreCount = reader.getInt(gpuCoreCountCol) // Default to 0 if not present
                 val gpuMemory = 0L // currently not implemented
+
+                val parents = reader.getSet(parentsCol, String::class.java) // No dependencies in the trace
+                val children = reader.getSet(childrenCol, String::class.java) // No dependencies in the trace
+
                 val uid = UUID.nameUUIDFromBytes("$id-${counter++}".toByteArray())
                 var nature = reader.getString(natureCol)
                 var deadline = reader.getLong(deadlineCol)
@@ -181,15 +187,17 @@ public class ComputeWorkloadLoader(
                     Task(
                         uid,
                         id,
+                        submissionTime,
+                        duration,
+                        parents!!,
+                        children!!,
                         cpuCount,
                         cpuCapacity,
+                        totalLoad,
                         memCapacity.roundToLong(),
                         gpuCoreCount,
                         gpuUsage,
                         gpuMemory,
-                        totalLoad,
-                        submissionTime,
-                        duration,
                         nature,
                         deadline,
                         builder.build(),

opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt

@@ -40,15 +40,17 @@ import java.util.UUID
 public data class Task(
     val uid: UUID,
     val name: String,
+    var submissionTime: Long,
+    val duration: Long,
+    val parents: Set<String> = emptySet(),
+    val children: Set<String> = emptySet(),
     val cpuCount: Int,
     val cpuCapacity: Double,
+    val totalCpuLoad: Double,
     val memCapacity: Long,
     val gpuCount: Int = 0,
     val gpuCapacity: Double = 0.0,
     val gpuMemCapacity: Long = 0L,
-    val totalLoad: Double,
-    var submissionTime: Long,
-    val duration: Long,
     val nature: String?,
     var deadline: Long,
     val trace: TraceWorkload,

opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/WorkloadLoader.kt

@@ -64,15 +64,15 @@ public abstract class WorkloadLoader(private val submissionTime: String? = null)
 
         val res = mutableListOf<Task>()
 
-        val totalLoad = workload.sumOf { it.totalLoad }
+        val totalLoad = workload.sumOf { it.totalCpuLoad }
         val desiredLoad = totalLoad * fraction
         var currentLoad = 0.0
 
         while (currentLoad < desiredLoad) {
             val entry = workload.random()
             res += entry
 
-            currentLoad += entry.totalLoad
+            currentLoad += entry.totalCpuLoad
         }
 
         logger.info { "Sampled ${workload.size} VMs (fraction $fraction) into subset of ${res.size} VMs" }

opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt

@@ -150,6 +150,8 @@ public suspend fun ComputeService.replay(
                                 entry.cpuCount,
                                 entry.memCapacity,
                                 entry.gpuCount,
+                                entry.parents,
+                                entry.children,
                                 flavorMeta,
                             ),
                             workload,

opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt

@@ -54,7 +54,6 @@ import java.time.LocalDateTime
 import java.time.ZoneOffset
 import java.util.UUID
 import kotlin.collections.ArrayList
-import kotlin.compareTo
 
 /**
  * Obtain the topology factory for the test.
@@ -86,15 +85,17 @@ fun createTestTask(
     return Task(
         UUID.nameUUIDFromBytes(name.toByteArray()),
         name,
+        LocalDateTime.parse(submissionTime).toInstant(ZoneOffset.UTC).toEpochMilli(),
+        duration,
+        emptySet(),
+        emptySet(),
         fragments.maxOf { it.cpuCoreCount() },
         fragments.maxOf { it.cpuUsage },
+        1800000.0,
         memCapacity,
         gpuCount = fragments.maxOfOrNull { it.gpuCoreCount() } ?: 0,
         gpuCapacity = fragments.maxOfOrNull { it.gpuUsage } ?: 0.0,
         gpuMemCapacity = fragments.maxOfOrNull { it.gpuMemoryUsage } ?: 0L,
-        1800000.0,
-        LocalDateTime.parse(submissionTime).toInstant(ZoneOffset.UTC).toEpochMilli(),
-        duration,
         "",
         -1,
         TraceWorkload(

opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/conv/ResourceColumns.kt

@@ -90,6 +90,18 @@ public val resourceGpuCapacity: String = "gpu_capacity"
 @JvmField
 public val resourceGpuMemCapacity: String = "gpu_mem_capacity"
 
+/**
+ * The parents of the resource that need to be completed before this resource can be used.
+ */
+@JvmField
+public val resourceParents: String = "parents"
+
+/**
+ * The children of the resource that cannot be started before this is completed.
+ */
+@JvmField
+public val resourceChildren: String = "children"
+
 /**
  * Nature of the task. Delayable, interruptible, etc.
  */