Made some changes to improve RAM of OpenDC (#318)

DanteNiewenhuis · web-flow · commit 5965492a9bab · 2025-03-18T22:55:50.000+01:00
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
@@ -34,7 +34,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.SplittableRandom;
 import java.util.UUID;
@@ -594,15 +593,16 @@ public Flavor findFlavor(@NotNull UUID id) {
         }
 
         @NotNull
-        public Flavor newFlavor(@NotNull String name, int cpuCount, long memorySize, @NotNull Map<String, ?> meta) {
+        public ServiceFlavor newFlavor(
+                @NotNull String name, int cpuCount, long memorySize, @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, meta);
 
-            service.flavorById.put(uid, flavor);
-            service.flavors.add(flavor);
+            //            service.flavorById.put(uid, flavor);
+            //            service.flavors.add(flavor);
 
             return flavor;
         }
@@ -642,18 +642,19 @@ public Image newImage(@NotNull String name, @NotNull Map<String, String> labels,
         @NotNull
         public ServiceTask newTask(
                 @NotNull String name,
-                @NotNull Flavor flavor,
+                @NotNull ServiceFlavor flavor,
                 @NotNull Workload workload,
                 @NotNull Map<String, ?> meta) {
             checkOpen();
 
             final ComputeService service = this.service;
             UUID uid = new UUID(service.clock.millis(), service.random.nextLong());
 
-            final ServiceFlavor internalFlavor =
-                    Objects.requireNonNull(service.flavorById.get(flavor.getUid()), "Unknown flavor");
+            //            final ServiceFlavor internalFlavor =
+            //                    Objects.requireNonNull(service.flavorById.get(flavor.getUid()), "Unknown flavor");
 
-            ServiceTask task = new ServiceTask(service, uid, name, internalFlavor, workload, meta);
+            //            ServiceTask task = new ServiceTask(service, uid, name, internalFlavor, workload, meta);
+            ServiceTask task = new ServiceTask(service, uid, name, flavor, workload, meta);
 
             service.taskById.put(uid, task);
 
diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceTask.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceTask.java
@@ -49,7 +49,7 @@ public class ServiceTask {
     private final UUID uid;
 
     private final String name;
-    private final ServiceFlavor flavor;
+    private ServiceFlavor flavor;
     public Workload workload;
 
     private Map<String, ?> meta; // TODO: remove this
@@ -181,6 +181,9 @@ public void delete() {
         }
         service.delete(this);
 
+        this.workload = null;
+        this.flavor = null;
+
         this.setState(TaskState.DELETED);
     }
 
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt
@@ -45,6 +45,7 @@ public class ComputeMonitorProvisioningStep(
             OutputFiles.POWER_SOURCE to true,
             OutputFiles.BATTERY to true,
         ),
+    private val printFrequency: Int? = null,
 ) : ProvisioningStep {
     override fun apply(ctx: ProvisioningContext): AutoCloseable {
         val service =
@@ -59,6 +60,7 @@ public class ComputeMonitorProvisioningStep(
                 exportInterval,
                 startTime,
                 filesToExport,
+                printFrequency,
             )
         return metricReader
     }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt
@@ -68,8 +68,9 @@ public fun registerComputeMonitor(
             OutputFiles.POWER_SOURCE to true,
             OutputFiles.BATTERY to true,
         ),
+    printFrequency: Int? = null,
 ): ProvisioningStep {
-    return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval, startTime, filesToExport)
+    return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval, startTime, filesToExport, printFrequency)
 }
 
 /**
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/ComputeMetricReader.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/ComputeMetricReader.kt
@@ -64,6 +64,7 @@ public class ComputeMetricReader(
             OutputFiles.BATTERY to true,
             OutputFiles.SERVICE to true,
         ),
+    private val printFrequency: Int? = null,
 ) : AutoCloseable {
     private val logger = KotlinLogging.logger {}
     private val scope = CoroutineScope(dispatcher.asCoroutineDispatcher())
@@ -156,6 +157,7 @@ public class ComputeMetricReader(
             }
 
             for (task in this.service.tasksToRemove) {
+                this.taskTableReaders.remove(task)
                 task.delete()
             }
             this.service.clearTasksToRemove()
@@ -197,7 +199,7 @@ public class ComputeMetricReader(
                 monitor.record(this.serviceTableReader.copy())
             }
 
-            if (loggCounter >= 24) {
+            if (printFrequency != null && loggCounter % printFrequency == 0) {
                 var loggString = "\n\t\t\t\t\tMetrics after ${now.toEpochMilli() / 1000 / 60 / 60} hours:\n"
                 loggString += "\t\t\t\t\t\tTasks Total: ${this.serviceTableReader.tasksTotal}\n"
                 loggString += "\t\t\t\t\t\tTasks Active: ${this.serviceTableReader.tasksActive}\n"
@@ -206,7 +208,6 @@ public class ComputeMetricReader(
                 loggString += "\t\t\t\t\t\tTasks Terminated: ${this.serviceTableReader.tasksTerminated}\n"
 
                 this.logger.warn { loggString }
-                loggCounter = 0
             }
         } catch (cause: Throwable) {
             this.logger.warn(cause) { "Exporter threw an Exception" }
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt
@@ -126,7 +126,7 @@ public class ComputeWorkloadLoader(
                     continue
                 }
 
-                val submissionTime = reader.getInstant(submissionTimeCol)!!
+                val submissionTime = reader.getInstant(submissionTimeCol)!!.toEpochMilli()
                 val duration = reader.getLong(durationCol)
                 val cpuCount = reader.getInt(cpuCountCol)
                 val cpuCapacity = reader.getDouble(cpuCapacityCol)
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
@@ -23,7 +23,6 @@
 package org.opendc.compute.workload
 
 import org.opendc.simulator.compute.workload.trace.TraceWorkload
-import java.time.Instant
 import java.util.UUID
 
 /**
@@ -45,7 +44,7 @@ public data class Task(
     val cpuCapacity: Double,
     val memCapacity: Long,
     val totalLoad: Double,
-    var submissionTime: Instant,
+    var submissionTime: Long,
     val duration: Long,
-    val trace: TraceWorkload,
+    var trace: TraceWorkload,
 )
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/WorkloadLoader.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/WorkloadLoader.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.compute.workload
 import mu.KotlinLogging
-import java.time.Instant
 import java.time.LocalDateTime
 import java.time.ZoneOffset
 
@@ -34,13 +33,13 @@ public abstract class WorkloadLoader(private val submissionTime: String? = null)
             return
         }
 
-        val workloadSubmissionTime = workload.minOf({ it.submissionTime }).toEpochMilli()
+        val workloadSubmissionTime = workload.minOf({ it.submissionTime })
         val submissionTimeLong = LocalDateTime.parse(submissionTime).toInstant(ZoneOffset.UTC).toEpochMilli()
 
         val timeShift = submissionTimeLong - workloadSubmissionTime
 
         for (task in workload) {
-            task.submissionTime = Instant.ofEpochMilli(task.submissionTime.toEpochMilli() + timeShift)
+            task.submissionTime += timeShift
         }
     }
 
@@ -54,10 +53,6 @@ public abstract class WorkloadLoader(private val submissionTime: String? = null)
 
         reScheduleTasks(workload)
 
-//        if (fraction >= 1.0) {
-//            return workload
-//        }
-
         if (fraction <= 0.0) {
             throw Error("The fraction of tasks to load cannot be 0.0 or lower")
         }
@@ -75,19 +70,6 @@ public abstract class WorkloadLoader(private val submissionTime: String? = null)
             currentLoad += entry.totalLoad
         }
 
-//        val shuffledWorkload = workload.shuffled()
-//        for (entry in shuffledWorkload) {
-//            val entryLoad = entry.totalLoad
-//
-//            // TODO: ask Sacheen
-//            if ((currentLoad + entryLoad) / totalLoad > fraction) {
-//                break
-//            }
-//
-//            currentLoad += entryLoad
-//            res += entry
-//        }
-
         logger.info { "Sampled ${workload.size} VMs (fraction $fraction) into subset of ${res.size} VMs" }
 
         return res
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/ExportModelSpec.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/ExportModelSpec.kt
@@ -37,6 +37,7 @@ public data class ExportModelSpec(
     val computeExportConfig: ComputeExportConfig = ComputeExportConfig.ALL_COLUMNS,
     val filesToExport: List<OutputFiles> = OutputFiles.entries.toList(),
     var filesToExportDict: MutableMap<OutputFiles, Boolean> = OutputFiles.entries.associateWith { false }.toMutableMap(),
+    var printFrequency: Int? = 24,
 ) {
     init {
         require(exportInterval > 0) { "The Export interval has to be higher than 0" }
diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioReplayer.kt
@@ -104,7 +104,7 @@ public suspend fun ComputeService.replay(
 
             for (entry in trace.sortedBy { it.submissionTime }) {
                 val now = clock.millis()
-                val start = entry.submissionTime.toEpochMilli()
+                val start = entry.submissionTime
 
                 // Set the simulationOffset based on the starting time of the first task
                 if (simulationOffset == Long.MIN_VALUE) {
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
@@ -93,9 +93,9 @@ public fun runScenario(
                     checkpointIntervalScaling,
                     scalingPolicy,
                 )
-            val workload = workloadLoader.sampleByLoad(scenario.workloadSpec.sampleFraction)
+            var workload = workloadLoader.sampleByLoad(scenario.workloadSpec.sampleFraction)
 
-            val startTimeLong = workload.minOf { it.submissionTime }.toEpochMilli()
+            val startTimeLong = workload.minOf { it.submissionTime }
             val startTime = Duration.ofMillis(startTimeLong)
 
             val topology = clusterTopology(scenario.topologySpec.pathToFile)
@@ -153,6 +153,7 @@ public fun addExportModel(
             Duration.ofSeconds(scenario.exportModelSpec.exportInterval),
             startTime,
             scenario.exportModelSpec.filesToExportDict,
+            scenario.exportModelSpec.printFrequency,
         ),
     )
 }
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
@@ -80,7 +80,7 @@ fun createTestTask(
         fragments.maxOf { it.cpuUsage },
         memCapacity,
         1800000.0,
-        LocalDateTime.parse(submissionTime).toInstant(ZoneOffset.UTC),
+        LocalDateTime.parse(submissionTime).toInstant(ZoneOffset.UTC).toEpochMilli(),
         duration,
         TraceWorkload(
             fragments,
@@ -108,7 +108,7 @@ fun runTest(
         val seed = 0L
         Provisioner(dispatcher, seed).use { provisioner ->
 
-            val startTimeLong = workload.minOf { it.submissionTime }.toEpochMilli()
+            val startTimeLong = workload.minOf { it.submissionTime }
             val startTime = Duration.ofMillis(startTimeLong)
 
             provisioner.runSteps(
@@ -121,7 +121,7 @@ fun runTest(
             service.setTasksExpected(workload.size)
             service.setMetricReader(provisioner.getMonitor())
 
-            service.replay(timeSource, workload, failureModelSpec = failureModelSpec)
+            service.replay(timeSource, ArrayDeque(workload), failureModelSpec = failureModelSpec)
         }
     }
 
diff --git a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
@@ -274,7 +274,7 @@ public class OpenDCRunner(
 //                    val vms = workload.resolve(workloadLoader, Random(seed))
 
                     val vms = workloadLoader.sampleByLoad(scenario.workload.samplingFraction)
-                    val startTime = vms.minOf { it.submissionTime }.toEpochMilli()
+                    val startTime = vms.minOf { it.submissionTime }
 
                     provisioner.runSteps(
                         setupComputeService(

Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ public class ServiceTask {`
`49`	`49`	`private final UUID uid;`
`50`	`50`
`51`	`51`	`private final String name;`
`52`		`- private final ServiceFlavor flavor;`
	`52`	`+ private ServiceFlavor flavor;`
`53`	`53`	`public Workload workload;`
`54`	`54`
`55`	`55`	`private Map<String, ?> meta; // TODO: remove this`
`@@ -181,6 +181,9 @@ public void delete() {`
`181`	`181`	`}`
`182`	`182`	`service.delete(this);`
`183`	`183`
	`184`	`+ this.workload = null;`
	`185`	`+ this.flavor = null;`
	`186`	`+`
`184`	`187`	`this.setState(TaskState.DELETED);`
`185`	`188`	`}`
`186`	`189`
Original file line number	Diff line number	Diff line change
`@@ -126,7 +126,7 @@ public class ComputeWorkloadLoader(`
`126`	`126`	`continue`
`127`	`127`	`}`
`128`	`128`
`129`		`- val submissionTime = reader.getInstant(submissionTimeCol)!!`
	`129`	`+ val submissionTime = reader.getInstant(submissionTimeCol)!!.toEpochMilli()`
`130`	`130`	`val duration = reader.getLong(durationCol)`
`131`	`131`	`val cpuCount = reader.getInt(cpuCountCol)`
`132`	`132`	`val cpuCapacity = reader.getDouble(cpuCapacityCol)`