Adds load shifting over time (#319)

* Start time shifting

* Existing experiments work with new columns

* Remove unused traces dir

* Update java to 21 LTS and jacoco to be compatible

* Minimal working timeshifting

* Timeshift scheduler linked as carbon receiver

* Add basic tests for timeshift scheduler

* Run spotless apply

* Modify tarce format tests to support new fields

* Change all mentions of java 19 to 21

* Add a deferAll option to workload to make all tasks deferrable

* Run spotless apply

* Copy traces from resources in web dockerfile

9 files changed, 246 insertions, 6 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 eb8d3377..69625306 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
@@ -25,6 +25,7 @@ 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;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -499,7 +500,6 @@ public final class ComputeService implements AutoCloseable {
             SimHost host = hv.getHost();
 
             // Remove request from queue
-            taskQueue.remove(req);
             tasksPending--;
 
             LOGGER.info("Assigned task {} to host {}", task, host);
@@ -642,6 +642,9 @@ public final class ComputeService implements AutoCloseable {
         @NotNull
         public ServiceTask newTask(
                 @NotNull String name,
+                @NotNull TaskNature nature,
+                @NotNull TemporalAmount duration,
+                @NotNull Long deadline,
                 @NotNull ServiceFlavor flavor,
                 @NotNull Workload workload,
                 @NotNull Map<String, ?> meta) {
@@ -652,9 +655,9 @@ public final class ComputeService implements AutoCloseable {
 
             //            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, flavor, workload, meta);
+
+            ServiceTask task = new ServiceTask(service, uid, name, nature, duration, deadline, 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
index dac65d67..4d5611a8 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
@@ -23,6 +23,7 @@
 package org.opendc.compute.simulator.service;
 
 import java.time.Instant;
+import java.time.temporal.TemporalAmount;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -49,6 +50,9 @@ public class ServiceTask {
     private final UUID uid;
 
     private final String name;
+    private final TaskNature nature;
+    private final TemporalAmount duration;
+    private final Long deadline;
     private ServiceFlavor flavor;
     public Workload workload;
 
@@ -68,12 +72,18 @@ public class ServiceTask {
             ComputeService service,
             UUID uid,
             String name,
+            TaskNature nature,
+            TemporalAmount duration,
+            Long deadline,
             ServiceFlavor flavor,
             Workload workload,
             Map<String, ?> meta) {
         this.service = service;
         this.uid = uid;
         this.name = name;
+        this.nature = nature;
+        this.duration = duration;
+        this.deadline = deadline;
         this.flavor = flavor;
         this.workload = workload;
         this.meta = meta;
@@ -87,6 +97,21 @@ public class ServiceTask {
     }
 
     @NotNull
+    public TaskNature getNature() {
+        return nature;
+    }
+
+    @NotNull
+    public TemporalAmount getDuration() {
+        return duration;
+    }
+
+    @NotNull
+    public Long getDeadline() {
+        return deadline;
+    }
+
+    @NotNull
     public String getName() {
         return name;
     }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/TaskNature.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/TaskNature.java
new file mode 100644
index 00000000..ffb49143
--- /dev/null
+++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/TaskNature.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2025 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.compute.simulator.service;
+
+public class TaskNature {
+
+    public final boolean deferrable;
+
+    public TaskNature(boolean deferrable) {
+        this.deferrable = deferrable;
+    }
+}
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
index a18856f8..1145c15b 100644
--- 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
@@ -29,6 +29,7 @@ import org.opendc.compute.simulator.telemetry.ComputeMonitor
 import org.opendc.compute.simulator.telemetry.OutputFiles
 import org.opendc.compute.topology.specs.ClusterSpec
 import org.opendc.compute.topology.specs.HostSpec
+import org.opendc.simulator.compute.power.CarbonReceiver
 import java.time.Duration
 
 /**
@@ -84,7 +85,8 @@ public fun registerComputeMonitor(
 public fun setupHosts(
     serviceDomain: String,
     specs: List<ClusterSpec>,
+    carbonReceivers: List<CarbonReceiver>,
     startTime: Long = 0L,
 ): ProvisioningStep {
-    return HostsProvisioningStep(serviceDomain, specs, startTime)
+    return HostsProvisioningStep(serviceDomain, specs, carbonReceivers, startTime)
 }
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 68e263f8..c347e866 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
@@ -29,6 +29,7 @@ import org.opendc.compute.topology.specs.ClusterSpec
 import org.opendc.compute.topology.specs.HostSpec
 import org.opendc.compute.topology.specs.createSimBatteryPolicy
 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.BatteryAggregator
 import org.opendc.simulator.compute.power.batteries.SimBattery
@@ -46,6 +47,7 @@ import org.opendc.simulator.engine.graph.FlowEdge
 public class HostsProvisioningStep internal constructor(
     private val serviceDomain: String,
     private val clusterSpecs: List<ClusterSpec>,
+    private val carbonReceivers: List<CarbonReceiver>,
     private val startTime: Long = 0L,
 ) : ProvisioningStep {
     override fun apply(ctx: ProvisioningContext): AutoCloseable {
@@ -75,6 +77,9 @@ public class HostsProvisioningStep internal constructor(
             if (carbonFragments != null) {
                 carbonModel = CarbonModel(engine, carbonFragments, startTime)
                 carbonModel.addReceiver(simPowerSource)
+                for (receiver in carbonReceivers) {
+                    carbonModel.addReceiver(receiver)
+                }
             }
 
             if (cluster.battery != null) {
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt
index 8f2369dc..b83eafdd 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt
@@ -31,6 +31,7 @@ import org.opendc.compute.simulator.scheduler.weights.CoreRamWeigher
 import org.opendc.compute.simulator.scheduler.weights.InstanceCountWeigher
 import org.opendc.compute.simulator.scheduler.weights.RamWeigher
 import org.opendc.compute.simulator.scheduler.weights.VCpuWeigher
+import java.time.InstantSource
 import java.util.SplittableRandom
 import java.util.random.RandomGenerator
 
@@ -45,13 +46,15 @@ public enum class ComputeSchedulerEnum {
     ProvisionedCoresInv,
     Random,
     TaskNumMemorizing,
+    Timeshift,
 }
 
 public fun createComputeScheduler(
     name: String,
     seeder: RandomGenerator,
+    clock: InstantSource,
 ): ComputeScheduler {
-    return createComputeScheduler(ComputeSchedulerEnum.valueOf(name.uppercase()), seeder)
+    return createComputeScheduler(ComputeSchedulerEnum.valueOf(name.uppercase()), seeder, clock)
 }
 
 /**
@@ -60,6 +63,7 @@ public fun createComputeScheduler(
 public fun createComputeScheduler(
     name: ComputeSchedulerEnum,
     seeder: RandomGenerator,
+    clock: InstantSource,
 ): ComputeScheduler {
     val cpuAllocationRatio = 1.0
     val ramAllocationRatio = 1.5
@@ -116,5 +120,13 @@ public fun createComputeScheduler(
                 filters = listOf(ComputeFilter(), VCpuFilter(cpuAllocationRatio), RamFilter(ramAllocationRatio)),
                 random = SplittableRandom(seeder.nextLong()),
             )
+        ComputeSchedulerEnum.Timeshift ->
+            TimeshiftScheduler(
+                filters = listOf(ComputeFilter(), VCpuFilter(cpuAllocationRatio), RamFilter(ramAllocationRatio)),
+                weighers = listOf(RamWeigher(multiplier = 1.0)),
+                windowSize = 168,
+                clock = clock,
+                random = SplittableRandom(seeder.nextLong()),
+            )
     }
 }
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt
index 832482eb..386e0d63 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt
@@ -79,7 +79,6 @@ public class FilterScheduler(
         }
 
         val task = req.task
-        val hosts = hosts
         val filteredHosts = hosts.filter { host -> filters.all { filter -> filter.test(host, task) } }
 
         val subset =
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 c48e6fb0..faa1be6b 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
@@ -88,6 +88,7 @@ public class MemorizingScheduler(
         taskloop@ for (req in iter) {
             if (req.isCancelled) {
                 iter.remove()
+                continue
             }
 
             for (chosenListIndex in minAvailableHost until hostsQueue.size) {
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/TimeshiftScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/TimeshiftScheduler.kt
new file mode 100644
index 00000000..a856f366
--- /dev/null
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/TimeshiftScheduler.kt
@@ -0,0 +1,161 @@
+/*
+ * Copyright (c) 2025 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.compute.simulator.scheduler
+
+import org.opendc.compute.simulator.scheduler.filters.HostFilter
+import org.opendc.compute.simulator.scheduler.weights.HostWeigher
+import org.opendc.compute.simulator.service.HostView
+import org.opendc.compute.simulator.service.ServiceTask
+import org.opendc.simulator.compute.power.CarbonModel
+import org.opendc.simulator.compute.power.CarbonReceiver
+import java.time.Instant
+import java.time.InstantSource
+import java.util.LinkedList
+import java.util.SplittableRandom
+import java.util.random.RandomGenerator
+import kotlin.math.min
+
+public class TimeshiftScheduler(
+    private val filters: List<HostFilter>,
+    private val weighers: List<HostWeigher>,
+    private val windowSize: Int,
+    private val clock: InstantSource,
+    private val subsetSize: Int = 1,
+    private val random: RandomGenerator = SplittableRandom(0),
+) : ComputeScheduler, CarbonReceiver {
+    /**
+     * The pool of hosts available to the scheduler.
+     */
+    private val hosts = mutableListOf<HostView>()
+
+    init {
+        require(subsetSize >= 1) { "Subset size must be one or greater" }
+    }
+
+    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
+
+    override fun addHost(host: HostView) {
+        hosts.add(host)
+    }
+
+    override fun removeHost(host: HostView) {
+        hosts.remove(host)
+    }
+
+    override fun select(iter: MutableIterator<SchedulingRequest>): SchedulingResult {
+        var result: SchedulingResult? = null
+        for (req in iter) {
+            if (req.isCancelled) {
+                iter.remove()
+                continue
+            }
+
+            val task = req.task
+
+            if (carbonIntensity > thresholdCarbonIntensity) {
+                if (task.nature.deferrable) {
+                    val currentTime = clock.instant()
+                    val estimatedCompletion = currentTime.plus(task.duration)
+                    val deadline = Instant.ofEpochMilli(task.deadline)
+                    if (estimatedCompletion.isBefore(deadline)) {
+                        // No need to schedule this task in a high carbon intensity period
+                        continue
+                    }
+                }
+            }
+
+            val filteredHosts = hosts.filter { host -> filters.all { filter -> filter.test(host, task) } }
+
+            val subset =
+                if (weighers.isNotEmpty()) {
+                    val filterResults = weighers.map { it.getWeights(filteredHosts, task) }
+                    val weights = DoubleArray(filteredHosts.size)
+
+                    for (fr in filterResults) {
+                        val min = fr.min
+                        val range = (fr.max - min)
+
+                        // Skip result if all weights are the same
+                        if (range == 0.0) {
+                            continue
+                        }
+
+                        val multiplier = fr.multiplier
+                        val factor = multiplier / range
+
+                        for ((i, weight) in fr.weights.withIndex()) {
+                            weights[i] += factor * (weight - min)
+                        }
+                    }
+
+                    weights.indices
+                        .asSequence()
+                        .sortedByDescending { weights[it] }
+                        .map { filteredHosts[it] }
+                        .take(subsetSize)
+                        .toList()
+                } else {
+                    filteredHosts
+                }
+
+            val maxSize = min(subsetSize, subset.size)
+            if (maxSize == 0) {
+                result = SchedulingResult(SchedulingResultType.FAILURE, null, req)
+                break
+            } else {
+                iter.remove()
+                result = SchedulingResult(SchedulingResultType.SUCCESS, subset[random.nextInt(maxSize)], req)
+                break
+            }
+        }
+
+        if (result == null) return SchedulingResult(SchedulingResultType.EMPTY)
+
+        return result
+    }
+
+    override fun removeTask(
+        task: ServiceTask,
+        host: HostView?,
+    ) {}
+
+    override fun updateCarbonIntensity(newCarbonIntensity: Double) {
+        this.carbonIntensity = newCarbonIntensity
+        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
+    }
+
+    override fun setCarbonModel(carbonModel: CarbonModel?) {}
+
+    override fun removeCarbonModel(carbonModel: CarbonModel?) {}
+}