simulator: Introduce SimResourceScheduler

This change introduces the SimResourceScheduler interface, which is a
generic interface for scheduling the coordination and synchronization
between resource providers and resource consumers.

This interface replaces the need for users to manually specify the clock
and coroutine context per resource provider.

9 files changed, 238 insertions, 42 deletions

diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
index 1bcaf45f..6ae04f27 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
@@ -22,12 +22,10 @@
 
 package org.opendc.simulator.resources
 
-import java.time.Clock
-
 /**
  * Abstract implementation of [SimResourceAggregator].
  */
-public abstract class SimAbstractResourceAggregator(private val clock: Clock) : SimResourceAggregator {
+public abstract class SimAbstractResourceAggregator(private val scheduler: SimResourceScheduler) : SimResourceAggregator {
     /**
      * This method is invoked when the resource consumer consumes resources.
      */
@@ -77,7 +75,7 @@ public abstract class SimAbstractResourceAggregator(private val clock: Clock) :
 
     protected val outputContext: SimResourceContext
         get() = context
-    private val context = object : SimAbstractResourceContext(0.0, clock, _output) {
+    private val context = object : SimAbstractResourceContext(0.0, scheduler, _output) {
         override val remainingWork: Double
             get() {
                 val now = clock.millis()
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt
index d2f585b1..c03bfad5 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt
@@ -31,9 +31,16 @@ import kotlin.math.min
  */
 public abstract class SimAbstractResourceContext(
     initialCapacity: Double,
-    override val clock: Clock,
+    private val scheduler: SimResourceScheduler,
     private val consumer: SimResourceConsumer
-) : SimResourceContext {
+) : SimResourceContext, SimResourceFlushable {
+
+    /**
+     * The clock of the context.
+     */
+    public override val clock: Clock
+        get() = scheduler.clock
+
     /**
      * The capacity of the resource.
      */
@@ -143,21 +150,12 @@ public abstract class SimAbstractResourceContext(
 
             flush(isIntermediate = true)
             doStop()
-        } catch (cause: Throwable) {
-            doFail(cause)
         } finally {
             isProcessing = false
         }
     }
 
-    /**
-     * Flush the current active resource consumption.
-     *
-     * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be
-     * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer
-     * will be asked to deliver a new command and is essentially interrupted.
-     */
-    public fun flush(isIntermediate: Boolean = false) {
+    override fun flush(isIntermediate: Boolean) {
         // Flush is no-op when the consumer is finished or not yet started
         if (state != SimResourceState.Active) {
             return
@@ -226,7 +224,7 @@ public abstract class SimAbstractResourceContext(
             return
         }
 
-        flush()
+        scheduler.schedule(this, isIntermediate = false)
     }
 
     override fun toString(): String = "SimAbstractResourceContext[capacity=$capacity]"
@@ -234,7 +232,7 @@ public abstract class SimAbstractResourceContext(
     /**
      * A flag to indicate that the resource is currently processing a command.
      */
-    protected var isProcessing: Boolean = false
+    private var isProcessing: Boolean = false
 
     /**
      * The current command that is being processed.
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt
index 5d550ad8..5665abd1 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt
@@ -22,12 +22,10 @@
 
 package org.opendc.simulator.resources
 
-import java.time.Clock
-
 /**
  * A [SimResourceAggregator] that distributes the load equally across the input resources.
  */
-public class SimResourceAggregatorMaxMin(clock: Clock) : SimAbstractResourceAggregator(clock) {
+public class SimResourceAggregatorMaxMin(scheduler: SimResourceScheduler) : SimAbstractResourceAggregator(scheduler) {
     private val consumers = mutableListOf<Input>()
 
     override fun doConsume(work: Double, limit: Double, deadline: Long) {
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
index 8128c98b..a76cb1e3 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
@@ -22,7 +22,6 @@
 
 package org.opendc.simulator.resources
 
-import java.time.Clock
 import kotlin.math.max
 import kotlin.math.min
 
@@ -31,7 +30,7 @@ import kotlin.math.min
  */
 public class SimResourceDistributorMaxMin(
     override val input: SimResourceProvider,
-    private val clock: Clock,
+    private val scheduler: SimResourceScheduler,
     private val listener: Listener? = null
 ) : SimResourceDistributor {
     override val outputs: Set<SimResourceProvider>
@@ -220,7 +219,7 @@ public class SimResourceDistributorMaxMin(
             }
         }
 
-        assert(deadline >= clock.millis()) { "Deadline already passed" }
+        assert(deadline >= scheduler.clock.millis()) { "Deadline already passed" }
 
         this.totalRequestedSpeed = totalRequestedSpeed
         this.totalRequestedWork = totalRequestedWork
@@ -337,7 +336,7 @@ public class SimResourceDistributorMaxMin(
     private inner class OutputContext(
         private val provider: OutputProvider,
         consumer: SimResourceConsumer
-    ) : SimAbstractResourceContext(provider.capacity, clock, consumer), Comparable<OutputContext> {
+    ) : SimAbstractResourceContext(provider.capacity, scheduler, consumer), Comparable<OutputContext> {
         /**
          * The current command that is processed by the vCPU.
          */
@@ -402,6 +401,8 @@ public class SimResourceDistributorMaxMin(
             }
         }
 
+        private var isProcessing: Boolean = false
+
         override fun interrupt() {
             // Prevent users from interrupting the CPU while it is constructing its next command, this will only lead
             // to infinite recursion.
@@ -409,10 +410,16 @@ public class SimResourceDistributorMaxMin(
                 return
             }
 
-            super.interrupt()
+            try {
+                isProcessing = false
 
-            // Force the scheduler to re-schedule
-            schedule()
+                super.interrupt()
+
+                // Force the scheduler to re-schedule
+                schedule()
+            } finally {
+                isProcessing = true
+            }
         }
 
         override fun compareTo(other: OutputContext): Int = allowedSpeed.compareTo(other.allowedSpeed)
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlushable.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlushable.kt
new file mode 100644
index 00000000..f6a1a42e
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlushable.kt
@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2021 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.simulator.resources
+
+/**
+ * An interface used by the [SimResourceScheduler] to flush the progress of resource consumer.
+ */
+public interface SimResourceFlushable {
+    /**
+     * Flush the current active resource consumption.
+     *
+     * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be
+     * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer
+     * will be asked to deliver a new command and is essentially interrupted.
+     */
+    public fun flush(isIntermediate: Boolean)
+}
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceScheduler.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceScheduler.kt
new file mode 100644
index 00000000..a228c47b
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceScheduler.kt
@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2021 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.simulator.resources
+
+import java.time.Clock
+
+/**
+ * A resource scheduler is responsible for scheduling the communication and synchronization between multiple resource
+ * providers and consumers.
+ *
+ * By centralizing the scheduling logic, updates of resources within a single system can be scheduled and tracked more
+ * efficiently, reducing the overall work needed per update.
+ */
+public interface SimResourceScheduler {
+    /**
+     * The [Clock] associated with this scheduler.
+     */
+    public val clock: Clock
+
+    /**
+     * Schedule a direct interrupt for the resource context represented by [flushable].
+     *
+     * @param flushable The resource context that needs to be flushed.
+     * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be
+     * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer
+     * will be asked to deliver a new command and is essentially interrupted.
+     */
+    public fun schedule(flushable: SimResourceFlushable, isIntermediate: Boolean = false)
+
+    /**
+     * Schedule an interrupt in the future for the resource context represented by [flushable].
+     *
+     * This method will override earlier calls to this method for the same [flushable].
+     *
+     * @param flushable The resource context that needs to be flushed.
+     * @param timestamp The timestamp when the interrupt should happen.
+     * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be
+     * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer
+     * will be asked to deliver a new command and is essentially interrupted.
+     */
+    public fun schedule(flushable: SimResourceFlushable, timestamp: Long, isIntermediate: Boolean = false)
+
+    /**
+     * Batch the execution of several interrupts into a single call.
+     *
+     * This method is useful if you want to propagate the start of multiple resources (e.g., CPUs) in a single update.
+     */
+    public fun batch(block: () -> Unit)
+}
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSchedulerTrampoline.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSchedulerTrampoline.kt
new file mode 100644
index 00000000..cdbb4a6c
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSchedulerTrampoline.kt
@@ -0,0 +1,95 @@
+/*
+ * Copyright (c) 2021 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.simulator.resources
+
+import org.opendc.utils.TimerScheduler
+import java.time.Clock
+import java.util.ArrayDeque
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * A [SimResourceScheduler] queues all interrupts that occur during execution to be executed after.
+ *
+ * @param clock The virtual simulation clock.
+ */
+public class SimResourceSchedulerTrampoline(context: CoroutineContext, override val clock: Clock) : SimResourceScheduler {
+    /**
+     * The [TimerScheduler] to actually schedule the interrupts.
+     */
+    private val timers = TimerScheduler<Any>(context, clock)
+
+    /**
+     * A flag to indicate that an interrupt is currently running already.
+     */
+    private var isRunning: Boolean = false
+
+    /**
+     * The queue of resources to be flushed.
+     */
+    private val queue = ArrayDeque<Pair<SimResourceFlushable, Boolean>>()
+
+    override fun schedule(flushable: SimResourceFlushable, isIntermediate: Boolean) {
+        queue.add(flushable to isIntermediate)
+
+        if (isRunning) {
+            return
+        }
+
+        flush()
+    }
+
+    override fun schedule(flushable: SimResourceFlushable, timestamp: Long, isIntermediate: Boolean) {
+        timers.startSingleTimerTo(flushable, timestamp) {
+            schedule(flushable, isIntermediate)
+        }
+    }
+
+    override fun batch(block: () -> Unit) {
+        val wasAlreadyRunning = isRunning
+        try {
+            isRunning = true
+            block()
+        } finally {
+            if (!wasAlreadyRunning) {
+                isRunning = false
+            }
+        }
+    }
+
+    /**
+     * Flush the scheduled queue.
+     */
+    private fun flush() {
+        val visited = mutableSetOf<SimResourceFlushable>()
+        try {
+            isRunning = true
+            while (queue.isNotEmpty()) {
+                val (flushable, isIntermediate) = queue.poll()
+                flushable.flush(isIntermediate)
+                visited.add(flushable)
+            }
+        } finally {
+            isRunning = false
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt
index fe569096..3277b889 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt
@@ -22,8 +22,6 @@
 
 package org.opendc.simulator.resources
 
-import org.opendc.utils.TimerScheduler
-import java.time.Clock
 import kotlin.math.ceil
 import kotlin.math.min
 
@@ -31,13 +29,11 @@ import kotlin.math.min
  * A [SimResourceSource] represents a source for some resource of type [R] that provides bounded processing capacity.
  *
  * @param initialCapacity The initial capacity of the resource.
- * @param clock The virtual clock to track simulation time.
  * @param scheduler The scheduler to schedule the interrupts.
  */
 public class SimResourceSource(
     initialCapacity: Double,
-    private val clock: Clock,
-    private val scheduler: TimerScheduler<Any>
+    private val scheduler: SimResourceScheduler
 ) : SimResourceProvider {
     /**
      * The current processing speed of the resource.
@@ -96,22 +92,21 @@ public class SimResourceSource(
     /**
      * Internal implementation of [SimResourceContext] for this class.
      */
-    private inner class Context(consumer: SimResourceConsumer) : SimAbstractResourceContext(capacity, clock, consumer) {
+    private inner class Context(consumer: SimResourceConsumer) : SimAbstractResourceContext(capacity, scheduler, consumer) {
         override fun onIdle(deadline: Long) {
             // Do not resume if deadline is "infinite"
             if (deadline != Long.MAX_VALUE) {
-                scheduler.startSingleTimerTo(this, deadline) { flush() }
+                scheduler.schedule(this, deadline)
             }
         }
 
         override fun onConsume(work: Double, limit: Double, deadline: Long) {
             val until = min(deadline, clock.millis() + getDuration(work, speed))
-
-            scheduler.startSingleTimerTo(this, until, ::flush)
+            scheduler.schedule(this, until)
         }
 
         override fun onFinish() {
-            scheduler.cancel(this)
+            cancel()
 
             ctx = null
 
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
index c796c251..5dc1e68d 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
@@ -23,14 +23,13 @@
 package org.opendc.simulator.resources
 
 import kotlinx.coroutines.*
-import java.time.Clock
 
 /**
  * A [SimResourceSwitch] implementation that switches resource consumptions over the available resources using max-min
  * fair sharing.
  */
 public class SimResourceSwitchMaxMin(
-    clock: Clock,
+    scheduler: SimResourceScheduler,
     private val listener: Listener? = null
 ) : SimResourceSwitch {
     private val _outputs = mutableSetOf<SimResourceProvider>()
@@ -49,13 +48,13 @@ public class SimResourceSwitchMaxMin(
     /**
      * The aggregator to aggregate the resources.
      */
-    private val aggregator = SimResourceAggregatorMaxMin(clock)
+    private val aggregator = SimResourceAggregatorMaxMin(scheduler)
 
     /**
      * The distributor to distribute the aggregated resources.
      */
     private val distributor = SimResourceDistributorMaxMin(
-        aggregator.output, clock,
+        aggregator.output, scheduler,
         object : SimResourceDistributorMaxMin.Listener {
             override fun onSliceFinish(
                 switch: SimResourceDistributor,