Added Fragment scaling (#296)

* Added maxCpuDemand to TraceWorkload, don't know if this will be needed so might remove later.

Updated SimTraceWorkload to properly handle creating checkpoints

Fixed a bug with the updatedConsumers in the FlowDistributor

Implemented a first version of scaling the runtime of fragments.

* small update

* updated tests to reflect the changes in the checkpointing model

* Updated the checkpointing tests to reflect the changes made

* updated wrapper-validation-action

* Applied spotless

9 files changed, 468 insertions, 69 deletions

diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt
index 87c7abe9..cf40d88d 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt
@@ -25,6 +25,9 @@ package org.opendc.experiments.base.experiment.specs
 import kotlinx.serialization.Serializable
 import org.opendc.compute.workload.ComputeWorkloadLoader
 import org.opendc.compute.workload.WorkloadLoader
+import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling
+import org.opendc.simulator.compute.workload.trace.scaling.PerfectScaling
+import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy
 import java.io.File
 
 /**
@@ -41,6 +44,7 @@ public data class WorkloadSpec(
     val type: WorkloadTypes,
     val sampleFraction: Double = 1.0,
     val submissionTime: String? = null,
+    val scalingPolicy: ScalingPolicyEnum = ScalingPolicyEnum.NoDelay,
 ) {
     public val name: String = File(pathToFile).nameWithoutExtension
 
@@ -56,11 +60,6 @@ public data class WorkloadSpec(
  * @constructor Create empty Workload types
  */
 public enum class WorkloadTypes {
-    /**
-     * Compute workload
-     *
-     * @constructor Create empty Compute workload
-     */
     ComputeWorkload,
 }
 
@@ -74,6 +73,7 @@ public fun getWorkloadLoader(
     checkpointInterval: Long,
     checkpointDuration: Long,
     checkpointIntervalScaling: Double,
+    scalingPolicy: ScalingPolicy,
 ): WorkloadLoader {
     return when (type) {
         WorkloadTypes.ComputeWorkload ->
@@ -83,6 +83,19 @@ public fun getWorkloadLoader(
                 checkpointInterval,
                 checkpointDuration,
                 checkpointIntervalScaling,
+                scalingPolicy,
             )
     }
 }
+
+public enum class ScalingPolicyEnum {
+    NoDelay,
+    Perfect,
+}
+
+public fun getScalingPolicy(scalingPolicyEnum: ScalingPolicyEnum): ScalingPolicy {
+    return when (scalingPolicyEnum) {
+        ScalingPolicyEnum.NoDelay -> NoDelayScaling()
+        ScalingPolicyEnum.Perfect -> PerfectScaling()
+    }
+}
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
index c9c2729d..56278bf2 100644
--- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
+++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt
@@ -33,6 +33,7 @@ import org.opendc.compute.simulator.service.ComputeService
 import org.opendc.compute.simulator.telemetry.parquet.ParquetComputeMonitor
 import org.opendc.compute.topology.clusterTopology
 import org.opendc.experiments.base.experiment.Scenario
+import org.opendc.experiments.base.experiment.specs.getScalingPolicy
 import org.opendc.experiments.base.experiment.specs.getWorkloadLoader
 import org.opendc.simulator.kotlin.runSimulation
 import java.io.File
@@ -80,6 +81,8 @@ public fun runScenario(
             val checkpointDuration = scenario.checkpointModelSpec?.checkpointDuration ?: 0L
             val checkpointIntervalScaling = scenario.checkpointModelSpec?.checkpointIntervalScaling ?: 1.0
 
+            val scalingPolicy = getScalingPolicy(scenario.workloadSpec.scalingPolicy)
+
             val workloadLoader =
                 getWorkloadLoader(
                     scenario.workloadSpec.type,
@@ -88,6 +91,7 @@ public fun runScenario(
                     checkpointInterval,
                     checkpointDuration,
                     checkpointIntervalScaling,
+                    scalingPolicy,
                 )
             val workload = workloadLoader.sampleByLoad(scenario.workloadSpec.sampleFraction)
 
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt
index 091f506a..895eee92 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt
@@ -26,7 +26,7 @@ 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.TraceFragment
+import org.opendc.simulator.compute.workload.trace.TraceFragment
 import java.util.ArrayList
 
 /**
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/ExperimentTest.kt
index 56850558..e271fce7 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/ExperimentTest.kt
@@ -26,7 +26,7 @@ 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.TraceFragment
+import org.opendc.simulator.compute.workload.trace.TraceFragment
 import java.util.ArrayList
 
 /**
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt
index 90737ab6..3231f533 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt
@@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
 import org.opendc.compute.workload.Task
 import org.opendc.experiments.base.experiment.specs.TraceBasedFailureModelSpec
-import org.opendc.simulator.compute.workload.TraceFragment
+import org.opendc.simulator.compute.workload.trace.TraceFragment
 import java.util.ArrayList
 
 /**
@@ -225,15 +225,12 @@ class FailuresAndCheckpointingTest {
     }
 
     /**
-     * Failure test 1: Single Task with checkpointing
+     * Checkpointing test 1: Single Task with checkpointing
      * In this test, a single task is scheduled that is interrupted by a failure after 5 min.
-     * Because there is no checkpointing, the full task has to be rerun.
-     *
-     * This means the final runtime is 20 minutes
+     * The system is using checkpointing, taking snapshots every minute.
      *
-     * When the task is running, it is using 50% of the cpu.
-     * This means that half of the time is active, and half is idle.
-     * When the task is failed, all time is idle.
+     * This means that after failure, only 6 minutes of the task is left.
+     * However, taking a snapshot takes 1 second, which means 9 seconds have to be added to the total runtime.
      */
     @Test
     fun testCheckpoints1() {
@@ -256,22 +253,20 @@ class FailuresAndCheckpointingTest {
 
         assertAll(
             { assertEquals((10 * 60000) + (9 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } },
-            { assertEquals(((10 * 30000)).toLong(), monitor.hostIdleTimes["H01"]?.sum()) { "Idle time incorrect" } },
-            { assertEquals(((10 * 30000) + (9 * 1000)).toLong(), monitor.hostActiveTimes["H01"]?.sum()) { "Active time incorrect" } },
-            { assertEquals((10 * 60 * 150.0) + (9 * 200.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } },
+            { assertEquals((10 * 60 * 150.0) + (9 * 150.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } },
         )
     }
 
     /**
-     * Failure test 2: Single Task with scaling checkpointing
+     * Checkpointing test 2: Single Task with checkpointing, higher cpu demand
      * In this test, a single task is scheduled that is interrupted by a failure after 5 min.
-     * Because there is no checkpointing, the full task has to be rerun.
+     * The system is using checkpointing, taking snapshots every minute.
      *
-     * This means the final runtime is 20 minutes
+     * This means that after failure, only 16 minutes of the task is left.
+     * However, taking a snapshot takes 1 second, which means 19 seconds have to be added to the total runtime.
      *
-     * When the task is running, it is using 50% of the cpu.
-     * This means that half of the time is active, and half is idle.
-     * When the task is failed, all time is idle.
+     * This is similar to the previous test, but the cpu demand of taking a snapshot is higher.
+     * The cpu demand of taking a snapshot is as high as the highest fragment
      */
     @Test
     fun testCheckpoints2() {
@@ -281,6 +276,86 @@ class FailuresAndCheckpointingTest {
                     name = "0",
                     fragments =
                         arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 2000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                        ),
+                    checkpointInterval = 60 * 1000L,
+                    checkpointDuration = 1000L,
+                ),
+            )
+
+        val topology = createTopology("single_1_2000.json")
+
+        val monitor = runTest(topology, workload)
+
+        assertAll(
+            { assertEquals((20 * 60000) + (19 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } },
+            {
+                assertEquals(
+                    (10 * 60 * 200.0) + (10 * 60 * 150.0) + (19 * 200.0),
+                    monitor.hostEnergyUsages["H01"]?.sum(),
+                ) { "Incorrect energy usage" }
+            },
+        )
+    }
+
+    /**
+     * Checkpointing test 3: Single Task with checkpointing, higher cpu demand
+     * In this test, a single task is scheduled that is interrupted by a failure after 5 min.
+     * The system is using checkpointing, taking snapshots every minute.
+     *
+     * This means that after failure, only 16 minutes of the task is left.
+     * However, taking a snapshot takes 1 second, which means 19 seconds have to be added to the total runtime.
+     *
+     * This is similar to the previous test, but the fragments are reversed
+     *
+     */
+    @Test
+    fun testCheckpoints3() {
+        val workload: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 2000.0, 1),
+                        ),
+                    checkpointInterval = 60 * 1000L,
+                    checkpointDuration = 1000L,
+                ),
+            )
+
+        val topology = createTopology("single_1_2000.json")
+
+        val monitor = runTest(topology, workload)
+
+        assertAll(
+            { assertEquals((20 * 60000) + (19 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } },
+            {
+                assertEquals(
+                    (10 * 60 * 200.0) + (10 * 60 * 150.0) + (19 * 200.0),
+                    monitor.hostEnergyUsages["H01"]?.sum(),
+                ) { "Incorrect energy usage" }
+            },
+        )
+    }
+
+    /**
+     * Checkpointing test 4: Single Task with scaling checkpointing
+     * In this test, checkpointing is used, with a scaling factor of 1.5
+     *
+     * This means that the interval between checkpoints starts at 1 min, but is multiplied by 1.5 every snapshot.
+     *
+     */
+    @Test
+    fun testCheckpoints4() {
+        val workload: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
                             TraceFragment(10 * 60 * 1000, 1000.0, 1),
                         ),
                     checkpointInterval = 60 * 1000L,
@@ -295,20 +370,18 @@ class FailuresAndCheckpointingTest {
 
         assertAll(
             { assertEquals((10 * 60000) + (4 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } },
-            { assertEquals(((10 * 30000)).toLong(), monitor.hostIdleTimes["H01"]?.sum()) { "Idle time incorrect" } },
-            { assertEquals(((10 * 30000) + (4 * 1000)).toLong(), monitor.hostActiveTimes["H01"]?.sum()) { "Active time incorrect" } },
-            { assertEquals((10 * 60 * 150.0) + (4 * 200.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } },
+            { assertEquals((10 * 60 * 150.0) + (4 * 150.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } },
         )
     }
 
     /**
-     * Checkpoint test 3: Single Task, single failure with checkpointing
+     * Checkpointing test 5: Single Task, single failure with checkpointing
      * In this test, a single task is scheduled that is interrupted by a failure after 5 min.
      * Because there is no checkpointing, the full task has to be rerun.
      *
      */
     @Test
-    fun testCheckpoints3() {
+    fun testCheckpoints5() {
         val workload: ArrayList<Task> =
             arrayListOf(
                 createTestTask(
@@ -336,23 +409,7 @@ class FailuresAndCheckpointingTest {
             { assertEquals((960 * 1000) + 5000, monitor.maxTimestamp) { "Total runtime incorrect" } },
             {
                 assertEquals(
-                    ((300 * 1000) + (296 * 500) + (360 * 500)).toLong(),
-                    monitor.hostIdleTimes["H01"]?.sum(),
-                ) { "Idle time incorrect" }
-            },
-            {
-                assertEquals(
-                    ((296 * 500) + 4000 + (360 * 500) + 5000).toLong(),
-                    monitor.hostActiveTimes["H01"]?.sum(),
-                ) { "Active time incorrect" }
-            },
-            { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(0)) { "Incorrect energy usage" } },
-            { assertEquals(6000.0, monitor.hostEnergyUsages["H01"]?.get(5)) { "Incorrect energy usage" } },
-            { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(10)) { "Incorrect energy usage" } },
-            {
-                assertEquals(
-                    (296 * 150.0) + (4 * 200.0) + (300 * 100.0) +
-                        (360 * 150.0) + (5 * 200.0),
+                    (665 * 150.0) + (300 * 100.0),
                     monitor.hostEnergyUsages["H01"]?.sum(),
                 ) { "Incorrect energy usage" }
             },
@@ -360,13 +417,13 @@ class FailuresAndCheckpointingTest {
     }
 
     /**
-     * Checkpoint test 4: Single Task, repeated failure with checkpointing
+     * Checkpointing test 6: Single Task, repeated failure with checkpointing
      * In this test, a single task is scheduled that is interrupted by a failure after 5 min.
      * Because there is no checkpointing, the full task has to be rerun.
      *
      */
     @Test
-    fun testCheckpoints4() {
+    fun testCheckpoints6() {
         val workload: ArrayList<Task> =
             arrayListOf(
                 createTestTask(
@@ -394,23 +451,7 @@ class FailuresAndCheckpointingTest {
             { assertEquals((22 * 60000) + 1000, monitor.maxTimestamp) { "Total runtime incorrect" } },
             {
                 assertEquals(
-                    ((10 * 60000) + (2 * 296 * 500) + (120 * 500)).toLong(),
-                    monitor.hostIdleTimes["H01"]?.sum(),
-                ) { "Idle time incorrect" }
-            },
-            {
-                assertEquals(
-                    ((2 * 296 * 500) + 8000 + (120 * 500) + 1000).toLong(),
-                    monitor.hostActiveTimes["H01"]?.sum(),
-                ) { "Active time incorrect" }
-            },
-            { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(0)) { "Incorrect energy usage" } },
-            { assertEquals(6000.0, monitor.hostEnergyUsages["H01"]?.get(5)) { "Incorrect energy usage" } },
-            { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(10)) { "Incorrect energy usage" } },
-            {
-                assertEquals(
-                    (2 * 296 * 150.0) + (8 * 200.0) + (600 * 100.0) +
-                        (120 * 150.0) + (200.0),
+                    (300 * 150.0) + (300 * 100.0) + (300 * 150.0) + (300 * 100.0) + (121 * 150.0),
                     monitor.hostEnergyUsages["H01"]?.sum(),
                 ) { "Incorrect energy usage" }
             },
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt
index 4a7c9341..3d733360 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt
@@ -26,7 +26,7 @@ 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.TraceFragment
+import org.opendc.simulator.compute.workload.trace.TraceFragment
 import java.util.ArrayList
 
 /**
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FragmentScalingTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FragmentScalingTest.kt
new file mode 100644
index 00000000..b0aa3555
--- /dev/null
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FragmentScalingTest.kt
@@ -0,0 +1,337 @@
+/*
+ * 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 org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling
+import org.opendc.simulator.compute.workload.trace.scaling.PerfectScaling
+import java.util.ArrayList
+
+/**
+ * Testing suite containing tests that specifically test the scaling of trace fragments
+ */
+class FragmentScalingTest {
+    /**
+     * Scaling test 1: A single fitting task
+     * In this test, a single task is scheduled that should fit the system.
+     * This means nothing will be delayed regardless of the scaling policy
+     */
+    @Test
+    fun testScaling1() {
+        val workloadNoDelay: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 2000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+            )
+
+        val workloadPerfect: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 2000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+            )
+        val topology = createTopology("single_1_2000.json")
+
+        val monitorNoDelay = runTest(topology, workloadNoDelay)
+        val monitorPerfect = runTest(topology, workloadPerfect)
+
+        assertAll(
+            { assertEquals(1200000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(1200000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(2000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } },
+        )
+    }
+
+    /**
+     * Scaling test 2: A single task with a single non-fitting fragment
+     * In this test, a single task is scheduled that should not fit.
+     * This means the Task is getting only 2000 Mhz while it was demanding 4000 Mhz
+     *
+     * For the NoDelay scaling policy, the task should take the planned 10 min.
+     * For the Perfect scaling policy, the task should be slowed down by 50% resulting in a runtime of 20 min.
+     */
+    @Test
+    fun testScaling2() {
+        val workloadNoDelay: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 4000.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+            )
+
+        val workloadPerfect: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 4000.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+            )
+        val topology = createTopology("single_1_2000.json")
+
+        val monitorNoDelay = runTest(topology, workloadNoDelay)
+        val monitorPerfect = runTest(topology, workloadPerfect)
+
+        assertAll(
+            { assertEquals(600000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(1200000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(4000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(4000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+        )
+    }
+
+    /**
+     * Scaling test 3: A single task that switches between fitting and not fitting
+     * In this test, a single task is scheduled has one fragment that does not fit
+     * This means the second fragment is getting only 2000 Mhz while it was demanding 4000 Mhz
+     *
+     * For the NoDelay scaling policy, the task should take the planned 30 min.
+     * For the Perfect scaling policy, the second fragment should be slowed down by 50% resulting in a runtime of 20 min,
+     * and a total runtime of 40 min.
+     */
+    @Test
+    fun testScaling3() {
+        val workloadNoDelay: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 4000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 1500.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+            )
+
+        val workloadPerfect: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 4000.0, 1),
+                            TraceFragment(10 * 60 * 1000, 1500.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+            )
+        val topology = createTopology("single_1_2000.json")
+
+        val monitorNoDelay = runTest(topology, workloadNoDelay)
+        val monitorPerfect = runTest(topology, workloadPerfect)
+
+        assertAll(
+            { assertEquals(1800000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(2400000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(4000.0, monitorNoDelay.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(4000.0, monitorPerfect.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(1500.0, monitorNoDelay.taskCpuDemands["0"]?.get(19)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(4000.0, monitorPerfect.taskCpuDemands["0"]?.get(19)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(1500.0, monitorNoDelay.taskCpuSupplied["0"]?.get(19)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(19)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(1500.0, monitorPerfect.taskCpuDemands["0"]?.get(29)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(1500.0, monitorPerfect.taskCpuSupplied["0"]?.get(29)) { "The cpu supplied to task 0 is incorrect" } },
+        )
+    }
+
+    /**
+     * Scaling test 4: Two tasks, that both fit
+     * In this test, two tasks are scheduled that both fit
+     *
+     * For both scaling policies, the tasks should run without delay.
+     */
+    @Test
+    fun testScaling4() {
+        val workloadNoDelay: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+                createTestTask(
+                    name = "1",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 3000.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+            )
+
+        val workloadPerfect: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 1000.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+                createTestTask(
+                    name = "1",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 3000.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+            )
+        val topology = createTopology("single_2_2000.json")
+
+        val monitorNoDelay = runTest(topology, workloadNoDelay)
+        val monitorPerfect = runTest(topology, workloadPerfect)
+
+        assertAll(
+            { assertEquals(600000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(600000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } },
+            { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(3000.0, monitorNoDelay.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } },
+            { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+            { assertEquals(3000.0, monitorPerfect.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } },
+            { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(3000.0, monitorNoDelay.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } },
+            { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+            { assertEquals(3000.0, monitorPerfect.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } },
+        )
+    }
+
+    /**
+     * Scaling test 5: Two tasks, that don't fit together
+     * In this test, two tasks are scheduled that do not fit together
+     * This means the Task_1 is getting only 2000 Mhz while it was demanding 4000 Mhz
+     *
+     * For the NoDelay scaling policy, the tasks should complete in 10 min
+     * For the Perfect scaling policy, task_1 is delayed while task_0 is still going.
+     * In the first 10 min (while Task_0 is still running), Task_1 is running at 50%.
+     * This means that after Task_0 is done, Task_1 still needs to run for 5 minutes, making the total runtime 15 min.
+     */
+    @Test
+    fun testScaling5() {
+        val workloadNoDelay: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 2000.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+                createTestTask(
+                    name = "1",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 4000.0, 1),
+                        ),
+                    scalingPolicy = NoDelayScaling(),
+                ),
+            )
+
+        val workloadPerfect: ArrayList<Task> =
+            arrayListOf(
+                createTestTask(
+                    name = "0",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 2000.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+                createTestTask(
+                    name = "1",
+                    fragments =
+                        arrayListOf(
+                            TraceFragment(10 * 60 * 1000, 4000.0, 1),
+                        ),
+                    scalingPolicy = PerfectScaling(),
+                ),
+            )
+        val topology = createTopology("single_2_2000.json")
+
+//        val monitorNoDelay = runTest(topology, workloadNoDelay)
+        val monitorPerfect = runTest(topology, workloadPerfect)
+
+//        assertAll(
+//            { assertEquals(600000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } },
+//            { assertEquals(900000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } },
+//
+//            { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+//            { assertEquals(3000.0, monitorNoDelay.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } },
+//            { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } },
+//            { assertEquals(3000.0, monitorPerfect.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } },
+//
+//            { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+//            { assertEquals(3000.0, monitorNoDelay.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } },
+//            { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } },
+//            { assertEquals(3000.0, monitorPerfect.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } },
+//        )
+    }
+}
diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt
index 6d80ce56..f9a20c68 100644
--- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt
+++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt
@@ -30,7 +30,7 @@ import org.opendc.compute.simulator.scheduler.filters.ComputeFilter
 import org.opendc.compute.simulator.scheduler.filters.RamFilter
 import org.opendc.compute.simulator.scheduler.filters.VCpuFilter
 import org.opendc.compute.workload.Task
-import org.opendc.simulator.compute.workload.TraceFragment
+import org.opendc.simulator.compute.workload.trace.TraceFragment
 import java.util.ArrayList
 
 class SchedulerTest {
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 eadd97e4..df45f374 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
@@ -43,8 +43,10 @@ import org.opendc.compute.topology.specs.ClusterSpec
 import org.opendc.compute.workload.Task
 import org.opendc.experiments.base.experiment.specs.FailureModelSpec
 import org.opendc.experiments.base.runner.replay
-import org.opendc.simulator.compute.workload.TraceFragment
-import org.opendc.simulator.compute.workload.TraceWorkload
+import org.opendc.simulator.compute.workload.trace.TraceFragment
+import org.opendc.simulator.compute.workload.trace.TraceWorkload
+import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling
+import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy
 import org.opendc.simulator.kotlin.runSimulation
 import java.time.Duration
 import java.time.LocalDateTime
@@ -69,6 +71,7 @@ fun createTestTask(
     checkpointInterval: Long = 0L,
     checkpointDuration: Long = 0L,
     checkpointIntervalScaling: Double = 1.0,
+    scalingPolicy: ScalingPolicy = NoDelayScaling(),
 ): Task {
     return Task(
         UUID.nameUUIDFromBytes(name.toByteArray()),
@@ -84,6 +87,7 @@ fun createTestTask(
             checkpointInterval,
             checkpointDuration,
             checkpointIntervalScaling,
+            scalingPolicy,
         ),
     )
 }