Simplified the WorkloadLoader into a single class that can be extended when new workload types are added (#294)

2 files changed, 33 insertions, 15 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 7f34c508..a3414054 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
@@ -23,9 +23,8 @@
 package org.opendc.experiments.base.experiment.specs
 
 import kotlinx.serialization.Serializable
-import org.opendc.compute.workload.ComputeWorkload
-import org.opendc.compute.workload.sampleByLoad
-import org.opendc.compute.workload.trace
+import org.opendc.compute.workload.ComputeWorkloadLoader
+import org.opendc.compute.workload.WorkloadLoader
 import java.io.File
 
 /**
@@ -61,12 +60,22 @@ public enum class WorkloadTypes {
 }
 
 /**
- *
- *TODO: move to separate file
- * @param type
+ * Create a workload loader for the given workload
  */
-public fun getWorkloadType(type: WorkloadTypes): ComputeWorkload {
+public fun getWorkloadLoader(
+    type: WorkloadTypes,
+    pathToFile: File,
+    checkpointInterval: Long,
+    checkpointDuration: Long,
+    checkpointIntervalScaling: Double,
+): WorkloadLoader {
     return when (type) {
-        WorkloadTypes.ComputeWorkload -> trace().sampleByLoad(1.0)
+        WorkloadTypes.ComputeWorkload ->
+            ComputeWorkloadLoader(
+                pathToFile,
+                checkpointInterval,
+                checkpointDuration,
+                checkpointIntervalScaling,
+            )
     }
 }
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 b6b255a6..cda43eb7 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
@@ -32,9 +32,8 @@ import org.opendc.compute.simulator.scheduler.createComputeScheduler
 import org.opendc.compute.simulator.service.ComputeService
 import org.opendc.compute.simulator.telemetry.parquet.ParquetComputeMonitor
 import org.opendc.compute.topology.clusterTopology
-import org.opendc.compute.workload.ComputeWorkloadLoader
 import org.opendc.experiments.base.experiment.Scenario
-import org.opendc.experiments.base.experiment.specs.getWorkloadType
+import org.opendc.experiments.base.experiment.specs.getWorkloadLoader
 import org.opendc.simulator.kotlin.runSimulation
 import java.io.File
 import java.time.Duration
@@ -81,16 +80,26 @@ public fun runScenario(
             val checkpointDuration = scenario.checkpointModelSpec?.checkpointDuration ?: 0L
             val checkpointIntervalScaling = scenario.checkpointModelSpec?.checkpointIntervalScaling ?: 1.0
 
+//            val workloadLoader =
+//                ComputeWorkloadLoader(
+//                    File(scenario.workloadSpec.pathToFile),
+//                    checkpointInterval,
+//                    checkpointDuration,
+//                    checkpointIntervalScaling,
+//                )
+//            val tasks = getWorkloadType(scenario.workloadSpec.type).resolve(workloadLoader, Random(seed))
+
             val workloadLoader =
-                ComputeWorkloadLoader(
+                getWorkloadLoader(
+                    scenario.workloadSpec.type,
                     File(scenario.workloadSpec.pathToFile),
                     checkpointInterval,
                     checkpointDuration,
                     checkpointIntervalScaling,
                 )
-            val tasks = getWorkloadType(scenario.workloadSpec.type).resolve(workloadLoader, Random(seed))
+            val workload = workloadLoader.load()
 
-            val startTimeLong = tasks.minOf { it.submissionTime }.toEpochMilli()
+            val startTimeLong = workload.minOf { it.submissionTime }.toEpochMilli()
             val startTime = Duration.ofMillis(startTimeLong)
 
             val topology = clusterTopology(scenario.topologySpec.pathToFile, Random(seed))
@@ -106,12 +115,12 @@ public fun runScenario(
             addExportModel(provisioner, serviceDomain, scenario, seed, startTime, scenario.id)
 
             val service = provisioner.registry.resolve(serviceDomain, ComputeService::class.java)!!
-            service.setTasksExpected(tasks.size)
+            service.setTasksExpected(workload.size)
             service.setMetricReader(provisioner.getMonitor())
 
             service.replay(
                 timeSource,
-                tasks,
+                workload,
                 failureModelSpec = scenario.failureModelSpec,
                 seed = seed,
             )