Shift task execution to the valley of carbon emissions (#323)

* Move shifted task execution to the valley of carbon emissions

* Remove error from compute service

* spotless apply

* Run scheduling cycle everytime carbon intensity updates

Author: sacheendra

opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java

@@ -55,6 +55,8 @@
 import org.opendc.compute.simulator.scheduler.SchedulingResultType;
 import org.opendc.compute.simulator.telemetry.ComputeMetricReader;
 import org.opendc.compute.simulator.telemetry.SchedulerStats;
+import org.opendc.simulator.compute.power.CarbonModel;
+import org.opendc.simulator.compute.power.CarbonReceiver;
 import org.opendc.simulator.compute.power.SimPowerSource;
 import org.opendc.simulator.compute.power.batteries.SimBattery;
 import org.opendc.simulator.compute.workload.Workload;
@@ -64,7 +66,7 @@
 /**
  * The {@link ComputeService} hosts the API implementation of the OpenDC Compute Engine.
  */
-public final class ComputeService implements AutoCloseable {
+public final class ComputeService implements AutoCloseable, CarbonReceiver {
     private static final Logger LOGGER = LoggerFactory.getLogger(ComputeService.class);
 
     /**
@@ -433,6 +435,14 @@ void delete(ServiceTask task) {
         taskById.remove(task.getUid());
     }
 
+    public void updateCarbonIntensity(double newCarbonIntensity) {
+        requestSchedulingCycle();
+    }
+
+    public void setCarbonModel(CarbonModel carbonModel) {}
+
+    public void removeCarbonModel(CarbonModel carbonModel) {}
+
     /**
      * Indicate that a new scheduling cycle is needed due to a change to the service's state.
      */
@@ -449,12 +459,12 @@ private void requestSchedulingCycle() {
      * Run a single scheduling iteration.
      */
     private void doSchedule() {
-
         for (Iterator<SchedulingRequest> iterator = taskQueue.iterator();
                 iterator.hasNext();
                 iterator = taskQueue.iterator()) {
             final SchedulingResult result = scheduler.select(iterator);
             if (result.getResultType() == SchedulingResultType.EMPTY) {
+
                 break;
             }
             final HostView hv = result.getHost();

opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt

@@ -80,6 +80,8 @@ public class HostsProvisioningStep internal constructor(
                 for (receiver in carbonReceivers) {
                     carbonModel.addReceiver(receiver)
                 }
+                val computeService = ctx.registry.resolve(serviceDomain, ComputeService::class.java)!!
+                carbonModel.addReceiver(computeService)
             }
 
             if (cluster.battery != null) {

opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt

@@ -47,6 +47,7 @@ public enum class ComputeSchedulerEnum {
     Random,
     TaskNumMemorizing,
     Timeshift,
+    TimeshiftNoPeak,
 }
 
 public fun createComputeScheduler(
@@ -128,5 +129,14 @@ public fun createComputeScheduler(
                 clock = clock,
                 random = SplittableRandom(seeder.nextLong()),
             )
+        ComputeSchedulerEnum.TimeshiftNoPeak ->
+            TimeshiftScheduler(
+                filters = listOf(ComputeFilter(), VCpuFilter(cpuAllocationRatio), RamFilter(ramAllocationRatio)),
+                weighers = listOf(RamWeigher(multiplier = 1.0)),
+                windowSize = 168,
+                clock = clock,
+                peakShift = false,
+                random = SplittableRandom(seeder.nextLong()),
+            )
     }
 }

opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/TimeshiftScheduler.kt

@@ -41,6 +41,7 @@ public class TimeshiftScheduler(
     private val windowSize: Int,
     private val clock: InstantSource,
     private val subsetSize: Int = 1,
+    private val peakShift: Boolean = true,
     private val random: RandomGenerator = SplittableRandom(0),
 ) : ComputeScheduler, CarbonReceiver {
     /**
@@ -54,11 +55,7 @@ public class TimeshiftScheduler(
 
     private val pastCarbonIntensities = LinkedList<Double>()
     private var carbonRunningSum = 0.0
-
-    private var carbonIntensity = 0.0
-
-//    private var lowerCarbonIntensity = 0.0
-    private var thresholdCarbonIntensity = 0.0
+    private var isLowCarbon = false
 
     override fun addHost(host: HostView) {
         hosts.add(host)
@@ -78,7 +75,7 @@ public class TimeshiftScheduler(
 
             val task = req.task
 
-            if (carbonIntensity > thresholdCarbonIntensity) {
+            if (!isLowCarbon) {
                 if (task.nature.deferrable) {
                     val currentTime = clock.instant()
                     val estimatedCompletion = currentTime.plus(task.duration)
@@ -146,13 +143,33 @@ public class TimeshiftScheduler(
     ) {}
 
     override fun updateCarbonIntensity(newCarbonIntensity: Double) {
-        this.carbonIntensity = newCarbonIntensity
+        val previousCarbonIntensity =
+            if (this.pastCarbonIntensities.isEmpty()) {
+                0.0
+            } else {
+                this.pastCarbonIntensities.last()
+            }
         this.pastCarbonIntensities.addLast(newCarbonIntensity)
         this.carbonRunningSum += newCarbonIntensity
         if (this.pastCarbonIntensities.size > this.windowSize) {
             this.carbonRunningSum -= this.pastCarbonIntensities.removeFirst()
         }
-        this.thresholdCarbonIntensity = this.carbonRunningSum / this.pastCarbonIntensities.size
+
+        val thresholdCarbonIntensity = this.carbonRunningSum / this.pastCarbonIntensities.size
+
+        if (!peakShift) {
+            isLowCarbon = newCarbonIntensity < thresholdCarbonIntensity
+            return
+        }
+
+        isLowCarbon = (
+            (newCarbonIntensity < thresholdCarbonIntensity) &&
+                (newCarbonIntensity > previousCarbonIntensity)
+        ) ||
+            (
+                (newCarbonIntensity < 1.2 * thresholdCarbonIntensity) &&
+                    isLowCarbon
+            )
     }
 
     override fun setCarbonModel(carbonModel: CarbonModel?) {}