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

2 files changed, 22 insertions, 5 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)