Rewritten the Carbon model (#260)

atlarge-research · Nov 3, 2024 · f3e578a · f3e578a
1 parent 6fa203b
commit f3e578a
Show file tree

Hide file tree

Showing 24 changed files with 276 additions and 223 deletions.
diff --git a/...dc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTrace.kt b/...dc-compute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTrace.kt
diff --git a/...pute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceLoader.kt b/...pute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceLoader.kt
@@ -22,6 +22,7 @@
 
 package org.opendc.compute.carbon
 
+import org.opendc.simulator.compute.power.CarbonFragmentNew
 import org.opendc.trace.Trace
 import org.opendc.trace.conv.CARBON_INTENSITY_TIMESTAMP
 import org.opendc.trace.conv.CARBON_INTENSITY_VALUE
@@ -40,14 +41,14 @@ public class CarbonTraceLoader {
     /**
      * The cache of workloads.
      */
-    private val cache = ConcurrentHashMap<String, SoftReference<List<CarbonFragment>>>()
+    private val cache = ConcurrentHashMap<String, SoftReference<List<CarbonFragmentNew>>>()
 
-    private val builder = CarbonFragmentBuilder()
+    private val builder = CarbonFragmentNewBuilder()
 
     /**
      * Read the metadata into a workload.
      */
-    private fun parseCarbon(trace: Trace): List<CarbonFragment> {
+    private fun parseCarbon(trace: Trace): List<CarbonFragmentNew> {
         val reader = checkNotNull(trace.getTable(TABLE_CARBON_INTENSITIES)).newReader()
 
         val startTimeCol = reader.resolve(CARBON_INTENSITY_TIMESTAMP)
@@ -76,7 +77,7 @@ public class CarbonTraceLoader {
     /**
      * Load the trace with the specified [name] and [format].
      */
-    public fun get(pathToFile: File): List<CarbonFragment> {
+    public fun get(pathToFile: File): List<CarbonFragmentNew> {
         val trace = Trace.open(pathToFile, "carbon")
 
         return parseCarbon(trace)
@@ -92,11 +93,11 @@ public class CarbonTraceLoader {
     /**
      * A builder for a VM trace.
      */
-    private class CarbonFragmentBuilder {
+    private class CarbonFragmentNewBuilder {
         /**
          * The total load of the trace.
          */
-        public val fragments: MutableList<CarbonFragment> = mutableListOf<CarbonFragment>()
+        public val fragments: MutableList<CarbonFragmentNew> = mutableListOf()
 
         /**
          * Add a fragment to the trace.
@@ -109,7 +110,7 @@ public class CarbonTraceLoader {
             carbonIntensity: Double,
         ) {
             fragments.add(
-                CarbonFragment(startTime.toEpochMilli(), Long.MAX_VALUE, carbonIntensity),
+                CarbonFragmentNew(startTime.toEpochMilli(), Long.MAX_VALUE, carbonIntensity),
             )
         }
 

diff --git a/...pute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceReader.kt b/...pute/opendc-compute-carbon/src/main/kotlin/org/opendc/compute/carbon/CarbonTraceReader.kt
@@ -20,33 +20,32 @@
  * SOFTWARE.
  */
 
-@file:JvmName("ComputeWorkloads")
+@file:JvmName("ComputeWorkloadsNew")
 
 package org.opendc.compute.carbon
 
+import org.opendc.simulator.compute.power.CarbonFragmentNew
 import java.io.File
 import javax.management.InvalidAttributeValueException
 
 /**
  * Construct a workload from a trace.
  */
-public fun getCarbonTrace(pathToFile: String?): CarbonTrace {
+public fun getCarbonFragments(pathToFile: String?): List<CarbonFragmentNew>? {
     if (pathToFile == null) {
-        return CarbonTrace(null)
+        return null
     }
 
-    return getCarbonTrace(File(pathToFile))
+    return getCarbonFragments(File(pathToFile))
 }
 
 /**
  * Construct a workload from a trace.
  */
-public fun getCarbonTrace(file: File): CarbonTrace {
+public fun getCarbonFragments(file: File): List<CarbonFragmentNew> {
     if (!file.exists()) {
         throw InvalidAttributeValueException("The carbon trace cannot be found")
     }
 
-    val fragments = CarbonTraceLoader().get(file)
-
-    return CarbonTrace(fragments)
+    return CarbonTraceLoader().get(file)
 }
diff --git a/...-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java b/...-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java
@@ -346,8 +346,11 @@ public void setTasksExpected(int numberOfTasks) {
 
     public void setTaskToBeRemoved(ServiceTask task) {
         this.tasksToRemove.add(task);
-        if ((tasksTerminated + tasksCompleted) == tasksExpected) {
-            metricReader.loggState(); // Logg the state for the final time. This will also delete all remaining tasks.
+        if ((this.tasksTerminated + this.tasksCompleted) == this.tasksExpected) {
+            if (this.metricReader != null) {
+                this.metricReader
+                        .loggState(); // Logg the state for the final time. This will also delete all remaining tasks.
+            }
         }
     }
 

diff --git a/...rc/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt b/...rc/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeMonitorProvisioningStep.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.compute.simulator.provisioner
 
-import org.opendc.compute.carbon.CarbonTrace
 import org.opendc.compute.simulator.service.ComputeService
 import org.opendc.compute.simulator.telemetry.ComputeMetricReader
 import org.opendc.compute.simulator.telemetry.ComputeMonitor
@@ -37,7 +36,6 @@ public class ComputeMonitorProvisioningStep(
     private val monitor: ComputeMonitor,
     private val exportInterval: Duration,
     private val startTime: Duration = Duration.ofMillis(0),
-    private val carbonTrace: CarbonTrace = CarbonTrace(null),
 ) : ProvisioningStep {
     override fun apply(ctx: ProvisioningContext): AutoCloseable {
         val service =
@@ -51,7 +49,6 @@ public class ComputeMonitorProvisioningStep(
                 monitor,
                 exportInterval,
                 startTime,
-                carbonTrace,
             )
         return metricReader
     }

diff --git a/...ompute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt b/...ompute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/ComputeSteps.kt
@@ -24,7 +24,6 @@
 
 package org.opendc.compute.simulator.provisioner
 
-import org.opendc.compute.carbon.CarbonTrace
 import org.opendc.compute.simulator.scheduler.ComputeScheduler
 import org.opendc.compute.simulator.telemetry.ComputeMonitor
 import org.opendc.compute.topology.specs.ClusterSpec
@@ -60,9 +59,8 @@ public fun registerComputeMonitor(
     monitor: ComputeMonitor,
     exportInterval: Duration = Duration.ofMinutes(5),
     startTime: Duration = Duration.ofMillis(0),
-    carbonTrace: CarbonTrace = CarbonTrace(null),
 ): ProvisioningStep {
-    return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval, startTime, carbonTrace)
+    return ComputeMonitorProvisioningStep(serviceDomain, monitor, exportInterval, startTime)
 }
 
 /**
@@ -76,6 +74,7 @@ public fun registerComputeMonitor(
 public fun setupHosts(
     serviceDomain: String,
     specs: List<ClusterSpec>,
+    startTime: Long = 0L,
 ): ProvisioningStep {
-    return HostsProvisioningStep(serviceDomain, specs)
+    return HostsProvisioningStep(serviceDomain, specs, startTime)
 }
diff --git a/...mulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt b/...mulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
@@ -22,6 +22,7 @@
 
 package org.opendc.compute.simulator.provisioner
 
+import org.opendc.compute.carbon.getCarbonFragments
 import org.opendc.compute.simulator.host.SimHost
 import org.opendc.compute.simulator.service.ComputeService
 import org.opendc.compute.topology.specs.ClusterSpec
@@ -40,6 +41,7 @@ import org.opendc.simulator.engine.FlowEngine
 public class HostsProvisioningStep internal constructor(
     private val serviceDomain: String,
     private val clusterSpecs: List<ClusterSpec>,
+    private val startTime: Long = 0L,
 ) : ProvisioningStep {
     override fun apply(ctx: ProvisioningContext): AutoCloseable {
         val service =
@@ -54,8 +56,11 @@ public class HostsProvisioningStep internal constructor(
 
         for (cluster in clusterSpecs) {
             // Create the Power Source to which hosts are connected
-            // TODO: Add connection to totalPower
-            val simPowerSource = SimPowerSource(graph)
+
+            val carbonFragments = getCarbonFragments(cluster.powerSource.carbonTracePath)
+
+            val simPowerSource = SimPowerSource(graph, cluster.powerSource.totalPower.toDouble(), carbonFragments, startTime)
+
             service.addPowerSource(simPowerSource)
             simPowerSources.add(simPowerSource)
 
@@ -88,7 +93,7 @@ public class HostsProvisioningStep internal constructor(
 
             for (simPowerSource in simPowerSources) {
                 // TODO: add close function
-//                simPowerSource.close()
+                simPowerSource.close()
             }
         }
     }

diff --git a/...e-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/ComputeMetricReader.kt b/...e-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/ComputeMetricReader.kt
@@ -29,7 +29,6 @@ import kotlinx.coroutines.launch
 import mu.KotlinLogging
 import org.opendc.common.Dispatcher
 import org.opendc.common.asCoroutineDispatcher
-import org.opendc.compute.carbon.CarbonTrace
 import org.opendc.compute.simulator.host.SimHost
 import org.opendc.compute.simulator.service.ComputeService
 import org.opendc.compute.simulator.service.ServiceTask
@@ -55,7 +54,6 @@ public class ComputeMetricReader(
     private val monitor: ComputeMonitor,
     private val exportInterval: Duration = Duration.ofMinutes(5),
     private val startTime: Duration = Duration.ofMillis(0),
-    private val carbonTrace: CarbonTrace = CarbonTrace(null),
 ) : AutoCloseable {
     private val logger = KotlinLogging.logger {}
     private val scope = CoroutineScope(dispatcher.asCoroutineDispatcher())
@@ -119,7 +117,6 @@ public class ComputeMetricReader(
                         HostTableReaderImpl(
                             it,
                             startTime,
-                            carbonTrace,
                         )
                     }
                 reader.record(now)
@@ -152,7 +149,6 @@ public class ComputeMetricReader(
                         PowerSourceTableReaderImpl(
                             it,
                             startTime,
-                            carbonTrace,
                         )
                     }
 

diff --git a/...r/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltHostExportColumns.kt b/...r/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltHostExportColumns.kt
@@ -154,16 +154,6 @@ public object DfltHostExportColumns {
             field = Types.required(FLOAT).named("energy_usage"),
         ) { it.energyUsage }
 
-    public val CARBON_INTENSITY: ExportColumn<HostTableReader> =
-        ExportColumn(
-            field = Types.required(FLOAT).named("carbon_intensity"),
-        ) { it.carbonIntensity }
-
-    public val CARBON_EMISSION: ExportColumn<HostTableReader> =
-        ExportColumn(
-            field = Types.required(FLOAT).named("carbon_emission"),
-        ) { it.carbonEmission }
-
     public val UP_TIME: ExportColumn<HostTableReader> =
         ExportColumn(
             field = Types.required(INT64).named("uptime"),

diff --git a/...simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/HostTableReader.kt b/...simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/HostTableReader.kt
@@ -122,16 +122,6 @@ public interface HostTableReader : Exportable {
      */
     public val energyUsage: Double
 
-    /**
-     * The current carbon intensity of the host in gCO2 / kW.
-     */
-    public val carbonIntensity: Double
-
-    /**
-     * The current carbon emission since the last deadline in g.
-     */
-    public val carbonEmission: Double
-
     /**
      * The uptime of the host since last time in ms.
      */