diff options
10 files changed, 666 insertions, 94 deletions
diff --git a/opendc-compute/opendc-compute-api/src/main/kotlin/org/opendc/compute/api/Flavor.kt b/opendc-compute/opendc-compute-api/src/main/kotlin/org/opendc/compute/api/Flavor.kt index 581c29ba..48765a3b 100644 --- a/opendc-compute/opendc-compute-api/src/main/kotlin/org/opendc/compute/api/Flavor.kt +++ b/opendc-compute/opendc-compute-api/src/main/kotlin/org/opendc/compute/api/Flavor.kt @@ -45,11 +45,6 @@ public interface Flavor : Resource { /** * Set of Tasks that need to be finished before this can startAdd commentMore actions */ - public val dependencies: Set<Int> - - /** - * Set of Tasks that need to be finished before this can startAdd commentMore actions - */ public val parents: Set<Int> /** diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java index 8feddf54..557fb760 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java +++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java @@ -111,7 +111,7 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver { */ private final Deque<SchedulingRequest> taskQueue = new ArrayDeque<>(); - private final List<SchedulingRequest> blockedTasks = new ArrayList<>(); + private final Map<Integer, SchedulingRequest> blockedTasks = new HashMap<>(); /** * The active tasks in the system. @@ -418,22 +418,10 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver { SchedulingRequest request = new SchedulingRequest(task, now); ServiceFlavor flavor = task.getFlavor(); - for (int taskId : this.terminatedTasks) { - if (flavor.isInDependencies(taskId)) { - // 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<Integer> pendingDependencies = flavor.getDependencies(); - if (!pendingDependencies.isEmpty()) { - // If the task has pending dependencies, we cannot schedule it yet - LOGGER.debug("Task {} has pending dependencies: {}", task.getId(), pendingDependencies); - blockedTasks.add(request); + // If the task has parents, put in blocked tasks + if (!flavor.getParents().isEmpty()) { + blockedTasks.put(task.getId(), request); return null; } @@ -447,51 +435,83 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver { return request; } - void addCompletedTask(ServiceTask task) { - int taskId = task.getId(); + void addCompletedTask(ServiceTask completedTask) { + int parentId = completedTask.getId(); + // int taskId = task.getId(); - if (!this.completedTasks.contains(taskId)) { - this.completedTasks.add(taskId); - } - - List<SchedulingRequest> requestsToRemove = new ArrayList<>(); + // if (!this.completedTasks.contains(taskId)) { + // this.completedTasks.add(taskId); + // } - for (SchedulingRequest request : blockedTasks) { - request.getTask().getFlavor().updatePendingDependencies(taskId); + for (int taskId : completedTask.getFlavor().getChildren()) { + SchedulingRequest request = blockedTasks.get(taskId); + if (request != null) { + request.getTask().getFlavor().removeFromParents(parentId); - Set<Integer> pendingDependencies = request.getTask().getFlavor().getDependencies(); + Set<Integer> pendingDependencies = request.getTask().getFlavor().getParents(); - if (pendingDependencies.isEmpty()) { - requestsToRemove.add(request); - taskQueue.add(request); - tasksPending++; + if (pendingDependencies.isEmpty()) { + taskQueue.add(request); + tasksPending++; + blockedTasks.remove(taskId); + } } } - for (SchedulingRequest request : requestsToRemove) { - blockedTasks.remove(request); - } + // List<SchedulingRequest> requestsToRemove = new ArrayList<>(); + // + // for (SchedulingRequest request : blockedTasks) { + // request.getTask().getFlavor().updatePendingDependencies(taskId); + // + // Set<Integer> 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) { - int taskId = task.getId(); - List<SchedulingRequest> requestsToRemove = new ArrayList<>(); + for (int taskId : task.getFlavor().getChildren()) { + SchedulingRequest request = blockedTasks.get(taskId); + if (request != null) { + ServiceTask childTask = request.getTask(); - if (!this.terminatedTasks.contains(taskId)) { - this.terminatedTasks.add(taskId); - } + childTask.setState(TaskState.TERMINATED); + + this.addTerminatedTask(childTask); - for (SchedulingRequest request : blockedTasks) { - if (request.getTask().getFlavor().isInDependencies(taskId)) { - requestsToRemove.add(request); - request.getTask().setState(TaskState.TERMINATED); + this.setTaskToBeRemoved(childTask); + + blockedTasks.remove(childTask.getId()); } } - for (SchedulingRequest request : requestsToRemove) { - blockedTasks.remove(request); - } + // int taskId = task.getId(); + // + // List<SchedulingRequest> requestsToRemove = new ArrayList<>(); + // + // if (!this.terminatedTasks.contains(taskId)) { + // this.terminatedTasks.add(taskId); + // } + // + // for (SchedulingRequest request : blockedTasks) { + // if (request.getTask().getFlavor().isInDependencies(taskId)) { + // requestsToRemove.add(request); + // request.getTask().setState(TaskState.TERMINATED); + // } + // } + // + // for (SchedulingRequest request : requestsToRemove) { + // blockedTasks.remove(request); + // } } void delete(ServiceFlavor flavor) { @@ -554,6 +574,8 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver { task.setState(TaskState.TERMINATED); + this.addTerminatedTask(task); + this.setTaskToBeRemoved(task); continue; } else { diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceFlavor.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceFlavor.java index 6201f21f..2d6f0342 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceFlavor.java +++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceFlavor.java @@ -23,7 +23,6 @@ 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; @@ -42,7 +41,6 @@ public final class ServiceFlavor implements Flavor { private final int gpuCoreCount; private final Set<Integer> parents; private final Set<Integer> children; - private final Set<Integer> dependencies; private final Map<String, ?> meta; ServiceFlavor( @@ -60,7 +58,6 @@ public final class ServiceFlavor implements Flavor { this.memorySize = memorySize; this.gpuCoreCount = gpuCoreCount; this.parents = parents; - this.dependencies = new HashSet<>(parents); this.children = children; this.meta = meta; } @@ -119,23 +116,18 @@ public final class ServiceFlavor implements Flavor { return "Flavor[name=" + taskId + "]"; } - @Override - public @NotNull Set<Integer> getDependencies() { - return dependencies; - } - - public void updatePendingDependencies(List<Integer> completedTasks) { + public void removeFromParents(List<Integer> completedTasks) { for (int task : completedTasks) { - this.updatePendingDependencies(task); + this.removeFromParents(task); } } - public void updatePendingDependencies(int completedTask) { - this.dependencies.remove(completedTask); + public void removeFromParents(int completedTask) { + this.parents.remove(completedTask); } public boolean isInDependencies(int task) { - return this.dependencies.contains(task); + return this.parents.contains(task); } @Override diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt index c875b8a2..a988d774 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt @@ -40,7 +40,7 @@ public data class Task( val name: String, var submissionTime: Long, val duration: Long, - val parents: Set<Int> = emptySet(), + val parents: Set<Int> = mutableSetOf(), val children: Set<Int> = emptySet(), val cpuCount: Int, val cpuCapacity: Double, diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentRunnerTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentCliTest.kt index cd16f174..748a020c 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentRunnerTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentCliTest.kt @@ -29,15 +29,15 @@ import java.io.File /** * An integration test suite for the Experiment Runner. */ -class ExperimentRunnerTest { +class ExperimentCliTest { /** - * ExperimentRunner test 1 + * ExperimentCli test 1 * This test runs the experiment defined in the experiment_1.json file. * * In this test, the bitbrains-small workload is executed with and without a carbon trace. */ @Test - fun testExperimentRunner1() { + fun testExperimentCli1() { ExperimentCommand().main(arrayOf("--experiment-path", "src/test/resources/experiments/experiment_1.json")) val someDir = File("output") @@ -45,13 +45,13 @@ class ExperimentRunnerTest { } /** - * ExperimentRunner test 2 + * ExperimentCli test 2 * This test runs the experiment defined in the experiment_2.json file. * * In this test, parts of the Marconi 100 workload is executed . This trace contains GPU tasks. */ @Test - fun testExperimentRunner2() { + fun testExperimentCli2() { ExperimentCommand().main(arrayOf("--experiment-path", "src/test/resources/experiments/experiment_2.json")) val someDir = File("output") diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioRunnerTest.kt index 28096bb8..f43dfbb0 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ScenarioRunnerTest.kt @@ -36,9 +36,9 @@ import java.util.ArrayList /** * An integration test suite for the Scenario experiments. */ -class ExperimentTest { +class ScenarioRunnerTest { /** - * Simulator test 1: Single Task + * Scenario test 1: Single Task * In this test, a single task is scheduled that takes 10 minutes to run. * * There should be no problems running the task, so the total runtime should be 10 min. @@ -48,7 +48,7 @@ class ExperimentTest { * When the task is failed, all time is idle. */ @Test - fun testSimulator1() { + fun testScenario1() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -76,7 +76,7 @@ class ExperimentTest { } /** - * Simulator test 1: Two Tasks + * Scenario test 1: Two Tasks * In this test, two tasks are scheduled. * * There should be no problems running the task, so the total runtime should be 15 min. @@ -85,7 +85,7 @@ class ExperimentTest { * The second task is using 100% of the available CPU capacity. */ @Test - fun testSimulator2() { + fun testScenario2() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -129,7 +129,7 @@ class ExperimentTest { } /** - * Simulator test 3: Two Tasks, one scheduled later + * Scenario test 3: Two Tasks, one scheduled later * In this test, two tasks are scheduled. * * There should be no problems running the task, so the total runtime should be 15 min. @@ -138,7 +138,7 @@ class ExperimentTest { * The second task is using 100% of the available CPU capacity. */ @Test - fun testSimulator3() { + fun testScenario3() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -174,7 +174,7 @@ class ExperimentTest { } /** - * Simulator test 4: Two Tasks, one scheduled later + * Scenario test 4: Two Tasks, one scheduled later * In this test, two tasks are scheduled. * * There should be no problems running the task, so the total runtime should be 15 min. @@ -183,7 +183,7 @@ class ExperimentTest { * The second task is using 100% of the available CPU capacity. */ @Test - fun testSimulator4() { + fun testScenario4() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -224,12 +224,12 @@ class ExperimentTest { } /** - * Simulator test 5: One Task purely running on GPU + * Scenario test 5: One Task purely running on GPU * * In this test, a single task is scheduled that takes 10 minutes to run. It solely uses the GPU. */ @Test - fun testSimulator5() { + fun testScenario5() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -272,12 +272,12 @@ class ExperimentTest { } /** - * Simulator test 6: One Task running on CPU & GPU + * Scenario test 6: One Task running on CPU & GPU * * In this test, a single task is scheduled that takes 10 minutes to run. CPU & GPU are used and have the same runtime. */ @Test - fun testSimulator6() { + fun testScenario6() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -319,12 +319,12 @@ class ExperimentTest { } /** - * Simulator test 7: One Task running on CPU & GPU + * Scenario test 7: One Task running on CPU & GPU * * In this test, a single task is scheduled that takes 10 minutes to run. CPU & GPU are used. CPU will finish way ahead of the GPU. */ @Test - fun testSimulator7() { + fun testScenario7() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -365,12 +365,12 @@ class ExperimentTest { } /** - * Simulator test 8: One Task running on CPU & GPU + * Scenario test 8: One Task running on CPU & GPU * * In this test, a single task is scheduled that takes 10 minutes to run. CPU & GPU are used. GPU will finish way ahead of the CPU. */ @Test - fun testSimulator8() { + fun testScenario8() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -410,12 +410,12 @@ class ExperimentTest { } /** - * Simulator test 9: Two tasks running on CPU & GPU + * Scenario test 9: Two tasks running on CPU & GPU * * In this test, two tasks are scheduled at the same time that takes 10 minutes to run. CPU & GPU are used. Both resources will finish at the same time. */ @Test - fun testSimulator9() { + fun testScenario9() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -465,12 +465,12 @@ class ExperimentTest { } /** - * Simulator test 10: Two tasks running on CPU & GPU + * Scenario test 10: Two tasks running on CPU & GPU * * In this test, two tasks are scheduled at the same time that takes 10 minutes to run. One task purely uses CPU, one purely GPU. */ @Test - fun testSimulator10() { + fun testScenario10() { val workload: ArrayList<Task> = arrayListOf( createTestTask( diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt index f34160f7..d5e4c925 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt @@ -75,6 +75,8 @@ fun createTestTask( checkpointDuration: Long = 0L, checkpointIntervalScaling: Double = 1.0, scalingPolicy: ScalingPolicy = NoDelayScaling(), + parents: Set<Int> = emptySet(), + children: Set<Int> = emptySet(), ): Task { var usedResources = arrayOf<ResourceType>() if (fragments.any { it.cpuUsage > 0.0 }) { @@ -89,8 +91,8 @@ fun createTestTask( name, LocalDateTime.parse(submissionTime).toInstant(ZoneOffset.UTC).toEpochMilli(), duration, - emptySet(), - emptySet(), + parents, + children, cpuCount, fragments.maxOf { it.cpuUsage }, 1800000.0, diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/WorkflowTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/WorkflowTest.kt new file mode 100644 index 00000000..6e149d45 --- /dev/null +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/WorkflowTest.kt @@ -0,0 +1,561 @@ +/* + * Copyright (c) 2020 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.experiments.base + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertAll +import org.opendc.compute.workload.Task +import org.opendc.simulator.compute.workload.trace.TraceFragment +import java.util.ArrayList + +/** + * An integration test suite for the Scenario experiments. + */ +class WorkflowTest { + /** + * Scenario test 1: simple workflow + * In this test, a simple workflow with 4 tasks is executed on a single host with 2 CPUs. + * The tasks are arranged in a diamond shape, where task 0 is the root task, tasks 1 and 2 are the children of task 0, + * and task 3 is the child of tasks 1 and 2. + * + * There should be no problems running the task, so the total runtime should be 30 min because 1 and 2 can be executed + * at the same time. + * + */ + @Test + fun testWorkflow1() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + id = 0, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = emptySet(), + children = mutableSetOf(1, 2), + ), + createTestTask( + id = 1, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 2, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 3, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(1, 2), + children = emptySet(), + ), + ) + + val topology = createTopology("single_2_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals(3 * 10 * 60 * 1000, monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + ((10 * 45000) + (10 * 30000) + (10 * 45000)).toLong(), + monitor.hostCpuIdleTimes["H01"]?.sum(), + ) { "Idle time incorrect" } + }, + { + assertEquals( + ((10 * 15000) + (10 * 30000) + (10 * 15000)).toLong(), + monitor.hostCpuActiveTimes["H01"]?.sum(), + ) { "Active time incorrect" } + }, + { assertEquals(7500.0, monitor.hostEnergyUsages["H01"]?.get(0)) { "Incorrect host energy usage at timestamp 0" } }, + { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(10)) { "Incorrect host energy usage at timestamp 0" } }, + { assertEquals(7500.0, monitor.hostEnergyUsages["H01"]?.get(20)) { "Incorrect host energy usage at timestamp 0" } }, + { + assertEquals( + 600 * 125.0 + 600 * 150.0 + 600 * 125.0, + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect host energy usage" } + }, + { assertEquals(600 * 125.0 + 600 * 150.0 + 600 * 125.0, monitor.energyUsages.sum()) { "Incorrect total energy usage" } }, + ) + } + + /** + * Scenario test 2: simple workflow + * In this test, a simple workflow with 4 tasks is executed on a single host with 2 CPUs. + * The tasks are arranged in a diamond shape, where task 0 is the root task, tasks 1 and 2 are the children of task 0, + * and task 3 is the child of tasks 1 and 2. However, task 2 has a shorter duration than task 1. + * + * There should be no problems running the task, so the total runtime should still be 30 min because + * 3 can only be executed after 1 is finished. + * + */ + @Test + fun testWorkflow2() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + id = 0, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = emptySet(), + children = mutableSetOf(1, 2), + ), + createTestTask( + id = 1, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 2, + fragments = + arrayListOf( + TraceFragment(5 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 3, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(1, 2), + children = emptySet(), + ), + ) + + val topology = createTopology("single_2_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals(3 * 10 * 60 * 1000, monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + ((10 * 45000) + (5 * 30000) + (15 * 45000)).toLong(), + monitor.hostCpuIdleTimes["H01"]?.sum(), + ) { "Idle time incorrect" } + }, + { + assertEquals( + ((10 * 15000) + (5 * 30000) + (15 * 15000)).toLong(), + monitor.hostCpuActiveTimes["H01"]?.sum(), + ) { "Active time incorrect" } + }, + { + assertEquals( + 7500.0, + monitor.hostEnergyUsages["H01"]?.get(0), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 9000.0, + monitor.hostEnergyUsages["H01"]?.get(10), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 7500.0, + monitor.hostEnergyUsages["H01"]?.get(20), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 600 * 125.0 + 300 * 150.0 + 900 * 125.0, + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect host energy usage" } + }, + { + assertEquals( + 600 * 125.0 + 300 * 150.0 + 900 * 125.0, + monitor.energyUsages.sum(), + ) { "Incorrect total energy usage" } + }, + ) + } + + /** + * Scenario test 3: simple workflow with unconnected extra task + * In this test, a simple workflow with 4 tasks is executed on a single host with 2 CPUs. + * However, there is also another task that is not connected to the workflow running. + * + * This means that the workflow cannot be parallelized as the extra task will take up one CPU for 40 minutes. + * + * The total runtime should therefore be 40 minutes. + */ + @Test + fun testWorkflow3() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + id = 0, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = emptySet(), + children = mutableSetOf(1, 2), + ), + createTestTask( + id = 1, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 2, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 3, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(1, 2), + children = emptySet(), + ), + createTestTask( + id = 5, + fragments = + arrayListOf( + TraceFragment(40 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + ), + ) + + val topology = createTopology("single_2_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals(4 * 10 * 60 * 1000, monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + ((40 * 30000)).toLong(), + monitor.hostCpuIdleTimes["H01"]?.sum(), + ) { "Idle time incorrect" } + }, + { + assertEquals( + ((40 * 30000)).toLong(), + monitor.hostCpuActiveTimes["H01"]?.sum(), + ) { "Active time incorrect" } + }, + { + assertEquals( + 9000.0, + monitor.hostEnergyUsages["H01"]?.get(0), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 9000.0, + monitor.hostEnergyUsages["H01"]?.get(10), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 9000.0, + monitor.hostEnergyUsages["H01"]?.get(20), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 2400 * 150.0, + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect host energy usage" } + }, + { + assertEquals( + 2400 * 150.0, + monitor.energyUsages.sum(), + ) { "Incorrect total energy usage" } + }, + ) + } + + /** + * Scenario test 4: simple workflow with unconnected extra task + * In this test, a simple workflow with 4 tasks is executed on a single host with 2 CPUs. + * However, there is also another task that is not connected to the workflow running. + * + * This means that the workflow cannot be parallelized for the first 15 minutes as the extra task will take up one CPU. + * After that, the workflow can be parallelized as the extra task is finished. + * + * The total runtime should therefore be 35 minutes. + */ + @Test + fun testWorkflow4() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + id = 0, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = emptySet(), + children = mutableSetOf(1, 2), + ), + createTestTask( + id = 1, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 2, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 3, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(1, 2), + children = emptySet(), + ), + createTestTask( + id = 5, + fragments = + arrayListOf( + TraceFragment(15 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + ), + ) + + val topology = createTopology("single_2_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals(35 * 60 * 1000, monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + ((20 * 30000) + (15 * 45000)).toLong(), + monitor.hostCpuIdleTimes["H01"]?.sum(), + ) { "Idle time incorrect" } + }, + { + assertEquals( + ((20 * 30000) + (15 * 15000)).toLong(), + monitor.hostCpuActiveTimes["H01"]?.sum(), + ) { "Active time incorrect" } + }, + { + assertEquals( + 9000.0, + monitor.hostEnergyUsages["H01"]?.get(0), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 9000.0, + monitor.hostEnergyUsages["H01"]?.get(10), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 7500.0, + monitor.hostEnergyUsages["H01"]?.get(20), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 1200 * 150.0 + 900 * 125.0, + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect host energy usage" } + }, + { + assertEquals( + 1200 * 150.0 + 900 * 125.0, + monitor.energyUsages.sum(), + ) { "Incorrect total energy usage" } + }, + ) + } + + /** + * Scenario test 3: workflow for which the first task cannot be scheduled. + * In this test, a simple workflow with 4 tasks is executed on a single host with 2 CPUs. + * However, the first task can not be scheduled on the host due to its high CPU requirement. + * The whole workflow is therefore terminated. + * + * The single unrelated task should still be executed + * + * The total runtime should therefore be 10 minutes. + */ + @Test + fun testWorkflow5() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + id = 0, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 10000.0), + ), + cpuCount = 10, + parents = emptySet(), + children = mutableSetOf(1, 2), + ), + createTestTask( + id = 1, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 2, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(0), + children = mutableSetOf(3), + ), + createTestTask( + id = 3, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + parents = mutableSetOf(1, 2), + children = emptySet(), + ), + createTestTask( + id = 5, + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0), + ), + cpuCount = 1, + ), + ) + + val topology = createTopology("single_2_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals(10 * 60 * 1000, monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + ((10 * 45000)).toLong(), + monitor.hostCpuIdleTimes["H01"]?.sum(), + ) { "Idle time incorrect" } + }, + { + assertEquals( + ((10 * 15000)).toLong(), + monitor.hostCpuActiveTimes["H01"]?.sum(), + ) { "Active time incorrect" } + }, + { + assertEquals( + 7500.0, + monitor.hostEnergyUsages["H01"]?.get(0), + ) { "Incorrect host energy usage at timestamp 0" } + }, + { + assertEquals( + 600 * 125.0, + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect host energy usage" } + }, + { + assertEquals( + 600 * 125.0, + monitor.energyUsages.sum(), + ) { "Incorrect total energy usage" } + }, + ) + } +} diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Task.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Task.kt index f661d5a9..0ebac5eb 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Task.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/Task.kt @@ -37,7 +37,7 @@ internal data class Task( val memCapacity: Double, val gpuCount: Int = 0, val gpuCapacity: Double = 0.0, - val parents: Set<Int> = emptySet(), + val parents: MutableSet<Int> = mutableSetOf(), val children: Set<Int> = emptySet(), val nature: String? = null, val deadline: Long = -1, diff --git a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskRecordMaterializer.kt b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskRecordMaterializer.kt index 12dc54b7..1b22e2a7 100644 --- a/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskRecordMaterializer.kt +++ b/opendc-trace/opendc-trace-api/src/main/kotlin/org/opendc/trace/formats/workload/parquet/TaskRecordMaterializer.kt @@ -170,7 +170,7 @@ internal class TaskRecordMaterializer(schema: MessageType) : RecordMaterializer< localMemCapacity, localGpuCount, localGpuCapacity, - localParents.toSet(), + localParents.toMutableSet(), localChildren.toSet(), localNature, localDeadline, |