Added embodied carbon to hosts (#326)

12 files changed, 86 insertions, 43 deletions

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
index 657e7f1e..b4b34881 100644
--- 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
@@ -23,7 +23,6 @@
 package org.opendc.compute.simulator.service;
 
 import java.time.Duration;
-import java.time.Instant;
 import java.time.InstantSource;
 import java.time.temporal.TemporalAmount;
 import java.util.ArrayDeque;
@@ -128,6 +127,7 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
 
     /**
      * The active tasks in the system.
+     * TODO: this is not doing anything, maybe delete it?
      */
     private final Map<ServiceTask, SimHost> completedTasks = new HashMap<>();
 
@@ -408,13 +408,30 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
      * Enqueue the specified [task] to be scheduled onto a host.
      */
     SchedulingRequest schedule(ServiceTask task) {
+        return schedule(task, false);
+    }
+
+    SchedulingRequest schedule(ServiceTask task, boolean atFront) {
         LOGGER.debug("Enqueueing task {} to be assigned to host", task.getUid());
 
+        if (task.getNumFailures() >= maxNumFailures) {
+            LOGGER.warn("task {} has been terminated because it failed {} times", task, task.getNumFailures());
+
+            tasksTerminated++;
+            task.setState(TaskState.TERMINATED);
+
+            this.setTaskToBeRemoved(task);
+            return null;
+        }
+
         long now = clock.millis();
         SchedulingRequest request = new SchedulingRequest(task, now);
 
-        task.scheduledAt = Instant.ofEpochMilli(now);
-        taskQueue.add(request);
+        if (atFront) {
+            taskQueue.addFirst(request);
+        } else {
+            taskQueue.add(request);
+        }
         tasksPending++;
         requestSchedulingCycle();
         return request;
@@ -473,18 +490,19 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
 
             final ServiceFlavor flavor = task.getFlavor();
 
-            if (task.getNumFailures() >= maxNumFailures) {
-                LOGGER.warn("task {} has been terminated because it failed {} times", task, task.getNumFailures());
-
-                taskQueue.remove(req);
-                tasksPending--;
-                tasksTerminated++;
-                task.setState(TaskState.TERMINATED);
-
-                scheduler.removeTask(task, hv);
-                this.setTaskToBeRemoved(task);
-                continue;
-            }
+            //            if (task.getNumFailures() >= maxNumFailures) {
+            //                LOGGER.warn("task {} has been terminated because it failed {} times", task,
+            // task.getNumFailures());
+            //
+            //                taskQueue.remove(req);
+            //                tasksPending--;
+            //                tasksTerminated++;
+            //                task.setState(TaskState.TERMINATED);
+            //
+            //                scheduler.removeTask(task, hv);
+            //                this.setTaskToBeRemoved(task);
+            //                continue;
+            //            }
 
             if (result.getResultType() == SchedulingResultType.FAILURE) {
                 LOGGER.trace("Task {} selected for scheduling but no capacity available for it at the moment", task);
@@ -516,6 +534,7 @@ public final class ComputeService implements AutoCloseable, CarbonReceiver {
 
             try {
                 task.host = host;
+                task.scheduledAt = clock.instant();
 
                 host.spawn(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
index 9fb9b5f0..cada796a 100644
--- 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
@@ -185,7 +185,7 @@ public class ServiceTask {
             case FAILED:
                 LOGGER.info("User requested to start task after failure {}", uid);
                 setState(TaskState.PROVISIONING);
-                request = service.schedule(this);
+                request = service.schedule(this, true);
                 break;
         }
     }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/telemetry/HostSystemStats.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/telemetry/HostSystemStats.java
index aa292797..8a7d022e 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/telemetry/HostSystemStats.java
+++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/telemetry/HostSystemStats.java
@@ -44,6 +44,7 @@ public record HostSystemStats(
         Instant bootTime,
         double powerDraw,
         double energyUsage,
+        double embodiedCarbon,
         int guestsTerminated,
         int guestsRunning,
         int guestsError,
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt
index 53fafe5f..9c58f7ab 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/host/SimHost.kt
@@ -46,7 +46,6 @@ import java.time.InstantSource
  *
  * @param name The name of the host.
  * @param clock The (virtual) clock used to track time.
- * @param graph The Flow Graph that the Host is part of
  * @param machineModel The static model of the host
  * @param cpuPowerModel The power model of the host
  * @param powerDistributor The power distributor to which the host is connected
@@ -59,6 +58,8 @@ public class SimHost(
     private val engine: FlowEngine,
     private val machineModel: MachineModel,
     private val cpuPowerModel: CpuPowerModel,
+    private val embodiedCarbon: Double,
+    private val expectedLifetime: Double,
     private val powerDistributor: FlowDistributor,
 ) : AutoCloseable {
     /**
@@ -109,6 +110,8 @@ public class SimHost(
     private var bootTime: Instant? = null
     private val cpuLimit = machineModel.cpuModel.totalCapacity
 
+    private var embodiedCarbonRate: Double = 0.0
+
     init {
         launch()
     }
@@ -117,6 +120,9 @@ public class SimHost(
      * Launch the hypervisor.
      */
     private fun launch() {
+        this.embodiedCarbonRate =
+            (this.embodiedCarbon * 1000) / (this.expectedLifetime * 365.0 * 24.0 * 60.0 * 60.0 * 1000.0)
+
         bootTime = this.clock.instant()
         hostState = HostState.UP
 
@@ -264,10 +270,12 @@ public class SimHost(
     }
 
     public fun getSystemStats(): HostSystemStats {
+        val now = clock.millis()
+        val duration = now - lastReport
         updateUptime()
         this.simMachine!!.psu.updateCounters()
 
-        var terminated = 0
+        val terminated = 0
         var running = 0
         var failed = 0
         var invalid = 0
@@ -293,6 +301,7 @@ public class SimHost(
             bootTime,
             simMachine!!.psu.powerDraw,
             simMachine!!.psu.energyUsage,
+            embodiedCarbonRate * duration,
             terminated,
             running,
             failed,
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
index d427b21c..75508b8d 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
@@ -133,6 +133,8 @@ public class HostsProvisioningStep internal constructor(
                         engine,
                         hostSpec.model,
                         hostSpec.cpuPowerModel,
+                        hostSpec.embodiedCarbon,
+                        hostSpec.expectedLifetime,
                         hostDistributor,
                     )
 
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt
index faa1be6b..4a561e95 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt
@@ -40,7 +40,7 @@ public class MemorizingScheduler(
 ) : ComputeScheduler {
     // We assume that there will be max 200 tasks per host.
     // The index of a host list is the number of tasks on that host.
-    private val hostsQueue = List(200, { mutableListOf<HostView>() })
+    private val hostsQueue = List(20000, { mutableListOf<HostView>() })
     private var minAvailableHost = 0
     private var numHosts = 0
 
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltHostExportColumns.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltHostExportColumns.kt
index f43f4b31..a691bc45 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltHostExportColumns.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltHostExportColumns.kt
@@ -154,6 +154,11 @@ public object DfltHostExportColumns {
             field = Types.required(FLOAT).named("energy_usage"),
         ) { it.energyUsage }
 
+    public val EMBODIED_CARBON: ExportColumn<HostTableReader> =
+        ExportColumn(
+            field = Types.required(FLOAT).named("embodied_carbon"),
+        ) { it.embodiedCarbon }
+
     public val UP_TIME: ExportColumn<HostTableReader> =
         ExportColumn(
             field = Types.required(INT64).named("uptime"),
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltTaskExportColumns.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltTaskExportColumns.kt
index 8603f669..2f9633db 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltTaskExportColumns.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/parquet/DfltTaskExportColumns.kt
@@ -66,14 +66,6 @@ public object DfltTaskExportColumns {
                     .named("task_id"),
         ) { Binary.fromString(it.taskInfo.id) }
 
-//    public val HOST_ID: ExportColumn<TaskTableReader> =
-//        ExportColumn(
-//            field =
-//                Types.optional(BINARY)
-//                    .`as`(LogicalTypeAnnotation.stringType())
-//                    .named("host_id"),
-//        ) { it.host?.id?.let { Binary.fromString(it) } }
-
     public val TASK_NAME: ExportColumn<TaskTableReader> =
         ExportColumn(
             field =
@@ -137,10 +129,10 @@ public object DfltTaskExportColumns {
             field = Types.required(INT64).named("downtime"),
         ) { it.downtime }
 
-    public val PROVISION_TIME: ExportColumn<TaskTableReader> =
+    public val NUM_FAILURES: ExportColumn<TaskTableReader> =
         ExportColumn(
-            field = Types.optional(INT64).named("provision_time"),
-        ) { it.provisionTime?.toEpochMilli() }
+            field = Types.required(INT64).named("num_failures"),
+        ) { it.numFailures }
 
     public val SCHEDULE_TIME: ExportColumn<TaskTableReader> =
         ExportColumn(
@@ -172,5 +164,7 @@ public object DfltTaskExportColumns {
         setOf(
             TIMESTAMP_ABS,
             TIMESTAMP,
+            TASK_ID,
+            TASK_NAME,
         )
 }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReader.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReader.kt
index 53282b9f..5cbdcd28 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReader.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReader.kt
@@ -123,6 +123,11 @@ public interface HostTableReader : Exportable {
     public val energyUsage: Double
 
     /**
+     * The embodied carbon emitted since the last sample in gram.
+     */
+    public val embodiedCarbon: Double
+
+    /**
      * The uptime of the host since last time in ms.
      */
     public val uptime: Long
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReaderImpl.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReaderImpl.kt
index 4990f0a3..913e0a17 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReaderImpl.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/host/HostTableReaderImpl.kt
@@ -59,6 +59,7 @@ public class HostTableReaderImpl(
         _cpuLostTime = table.cpuLostTime
         _powerDraw = table.powerDraw
         _energyUsage = table.energyUsage
+        _embodiedCarbon = table.embodiedCarbon
         _uptime = table.uptime
         _downtime = table.downtime
         _bootTime = table.bootTime
@@ -143,6 +144,10 @@ public class HostTableReaderImpl(
     private var _energyUsage = 0.0
     private var previousEnergyUsage = 0.0
 
+    override val embodiedCarbon: Double
+        get() = _embodiedCarbon
+    private var _embodiedCarbon = 0.0
+
     override val uptime: Long
         get() = _uptime - previousUptime
     private var _uptime = 0L
@@ -181,6 +186,7 @@ public class HostTableReaderImpl(
         _cpuLostTime = hostCpuStats.lostTime
         _powerDraw = hostSysStats.powerDraw
         _energyUsage = hostSysStats.energyUsage
+        _embodiedCarbon = hostSysStats.embodiedCarbon
         _uptime = hostSysStats.uptime.toMillis()
         _downtime = hostSysStats.downtime.toMillis()
         _bootTime = hostSysStats.bootTime
@@ -212,5 +218,6 @@ public class HostTableReaderImpl(
 
         _powerDraw = 0.0
         _energyUsage = 0.0
+        _embodiedCarbon = 0.0
     }
 }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReader.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReader.kt
index 85e030aa..79c8e4a6 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReader.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReader.kt
@@ -71,9 +71,9 @@ public interface TaskTableReader : Exportable {
     public val downtime: Long
 
     /**
-     * The [Instant] at which the task was enqueued for the scheduler.
+     * The number of times the task has been kicked from a host due to failures
      */
-    public val provisionTime: Instant?
+    public val numFailures: Int
 
     /**
      * The [Instant] at which the task was scheduled relative to the start of the workload.
@@ -86,7 +86,7 @@ public interface TaskTableReader : Exportable {
     public val submissionTime: Instant?
 
     /**
-     * The [Instant] at which the task booted relative to the start of the workload.
+     * The [Instant] at which the task finished relative to the start of the workload.
      */
     public val finishTime: Instant?
 
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReaderImpl.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReaderImpl.kt
index d8c6a06e..be8d5725 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReaderImpl.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/telemetry/table/task/TaskTableReaderImpl.kt
@@ -64,7 +64,7 @@ public class TaskTableReaderImpl(
         _cpuLostTime = table.cpuLostTime
         _uptime = table.uptime
         _downtime = table.downtime
-        _provisionTime = table.provisionTime
+        _numFailures = table.numFailures
         _scheduleTime = table.scheduleTime
 
         _submissionTime = table.submissionTime
@@ -110,18 +110,18 @@ public class TaskTableReaderImpl(
     private var _downtime: Long = 0
     private var previousDowntime = 0L
 
-    override val provisionTime: Instant?
-        get() = _provisionTime
-    private var _provisionTime: Instant? = null
-
-    override val scheduleTime: Instant?
-        get() = _scheduleTime
-    private var _scheduleTime: Instant? = null
+    override val numFailures: Int
+        get() = _numFailures
+    private var _numFailures = 0
 
     override val submissionTime: Instant?
         get() = _submissionTime
     private var _submissionTime: Instant? = null
 
+    override val scheduleTime: Instant?
+        get() = _scheduleTime
+    private var _scheduleTime: Instant? = null
+
     override val finishTime: Instant?
         get() = _finishTime
     private var _finishTime: Instant? = null
@@ -195,9 +195,10 @@ public class TaskTableReaderImpl(
         _cpuLostTime = cpuStats?.lostTime ?: _cpuLostTime
         _uptime = sysStats?.uptime?.toMillis() ?: _uptime
         _downtime = sysStats?.downtime?.toMillis() ?: _downtime
-        _provisionTime = task.scheduledAt
-        _scheduleTime = sysStats?.bootTime ?: _scheduleTime
+
+        _numFailures = task.numFailures
         _submissionTime = task.submittedAt
+        _scheduleTime = task.scheduledAt
         _finishTime = task.finishedAt
 
         _taskState = task.state