simulator: Centralize resource logic in SimResourceInterpreter

This change introduces the SimResourceInterpreter which centralizes the
logic for scheduling and interpreting the communication between resource
consumer and provider.

This approach offers better performance due to avoiding invalidating the
state of the resource context when not necessary. Benchmarks show in the
best case a 5x performance improvement and at worst a 2x improvement.

8 files changed, 64 insertions, 82 deletions

diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
index 713376e7..2df307d3 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
@@ -31,12 +31,13 @@ import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.*
+import org.opendc.simulator.resources.SimResourceSwitch
 import java.time.Clock
 
 /**
  * Abstract implementation of the [SimHypervisor] interface.
  */
-public abstract class SimAbstractHypervisor : SimHypervisor {
+public abstract class SimAbstractHypervisor(private val interpreter: SimResourceInterpreter) : SimHypervisor {
     /**
      * The machine on which the hypervisor runs.
      */
@@ -122,11 +123,13 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
                     override val meta: Map<String, Any> = meta
                 }
 
-                workload.onStart(ctx)
+                interpreter.batch {
+                    workload.onStart(ctx)
 
-                for (cpu in cpus) {
-                    launch {
-                        cpu.consume(workload.getConsumer(ctx, cpu.model))
+                    for (cpu in cpus) {
+                        launch {
+                            cpu.consume(workload.getConsumer(ctx, cpu.model))
+                        }
                     }
                 }
             }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
index e501033a..de2b3eef 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
@@ -27,15 +27,18 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.workload.SimWorkload
+import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.resources.SimResourceSystem
+import org.opendc.simulator.resources.batch
 import org.opendc.simulator.resources.consume
-import org.opendc.simulator.resources.consumer.SimSpeedConsumerAdapter
 import java.time.Clock
-import kotlin.coroutines.CoroutineContext
 
 /**
  * Abstract implementation of the [SimMachine] interface.
+ *
+ * @param interpreter The interpreter to manage the machine's resources.
  */
-public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine {
+public abstract class SimAbstractMachine(protected val interpreter: SimResourceInterpreter) : SimMachine, SimResourceSystem {
     private val _usage = MutableStateFlow(0.0)
     override val usage: StateFlow<Double>
         get() = _usage
@@ -53,11 +56,6 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
     private var isTerminated = false
 
     /**
-     * The [CoroutineContext] to run in.
-     */
-    protected abstract val context: CoroutineContext
-
-    /**
      * The resources allocated for this machine.
      */
     protected abstract val cpus: List<SimProcessingUnit>
@@ -67,7 +65,7 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
      */
     private inner class Context(override val meta: Map<String, Any>) : SimMachineContext {
         override val clock: Clock
-            get() = this@SimAbstractMachine.clock
+            get() = interpreter.clock
 
         override val cpus: List<SimProcessingUnit> = this@SimAbstractMachine.cpus
 
@@ -77,38 +75,35 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
     /**
      * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
      */
-    override suspend fun run(workload: SimWorkload, meta: Map<String, Any>): Unit = withContext(context) {
-        require(!isTerminated) { "Machine is terminated" }
+    override suspend fun run(workload: SimWorkload, meta: Map<String, Any>): Unit = coroutineScope {
+        check(!isTerminated) { "Machine is terminated" }
         val ctx = Context(meta)
-        val totalCapacity = model.cpus.sumOf { it.frequency }
-
-        _speed = DoubleArray(model.cpus.size) { 0.0 }
-        var totalSpeed = 0.0
 
         // Before the workload starts, initialize the initial power draw
+        _speed = DoubleArray(model.cpus.size) { 0.0 }
         updateUsage(0.0)
 
-        workload.onStart(ctx)
+        interpreter.batch {
+            workload.onStart(ctx)
 
-        for (cpu in cpus) {
-            val model = cpu.model
-            val consumer = workload.getConsumer(ctx, model)
-            val adapter = SimSpeedConsumerAdapter(consumer) { newSpeed ->
-                val _speed = _speed
-                val _usage = _usage
-
-                val oldSpeed = _speed[model.id]
-                _speed[model.id] = newSpeed
-                totalSpeed = totalSpeed - oldSpeed + newSpeed
-
-                val newUsage = totalSpeed / totalCapacity
-                if (_usage.value != newUsage) {
-                    updateUsage(totalSpeed / totalCapacity)
-                }
+            for (cpu in cpus) {
+                val model = cpu.model
+                val consumer = workload.getConsumer(ctx, model)
+                launch { cpu.consume(consumer) }
             }
+        }
+    }
 
-            launch { cpu.consume(adapter) }
+    override fun onConverge(timestamp: Long) {
+        val totalCapacity = model.cpus.sumOf { it.frequency }
+        val cpus = cpus
+        var totalSpeed = 0.0
+        for (cpu in cpus) {
+            _speed[cpu.model.id] = cpu.speed
+            totalSpeed += cpu.speed
         }
+
+        updateUsage(totalSpeed / totalCapacity)
     }
 
     /**
@@ -119,9 +114,11 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
     }
 
     override fun close() {
-        if (!isTerminated) {
-            isTerminated = true
-            cpus.forEach(SimProcessingUnit::close)
+        if (isTerminated) {
+            return
         }
+
+        isTerminated = true
+        cpus.forEach(SimProcessingUnit::close)
     }
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
index 27ebba21..f5218ba9 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
@@ -27,8 +27,7 @@ import org.opendc.simulator.compute.cpufreq.ScalingDriver
 import org.opendc.simulator.compute.cpufreq.ScalingGovernor
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.resources.*
-import org.opendc.utils.TimerScheduler
-import java.time.Clock
+import org.opendc.simulator.resources.SimResourceInterpreter
 import kotlin.coroutines.*
 
 /**
@@ -43,24 +42,11 @@ import kotlin.coroutines.*
  */
 @OptIn(ExperimentalCoroutinesApi::class, InternalCoroutinesApi::class)
 public class SimBareMetalMachine(
-    context: CoroutineContext,
-    private val clock: Clock,
+    platform: SimResourceInterpreter,
     override val model: SimMachineModel,
     scalingGovernor: ScalingGovernor,
     scalingDriver: ScalingDriver
-) : SimAbstractMachine(clock) {
-    /**
-     * The [Job] associated with this machine.
-     */
-    private val scope = CoroutineScope(context + Job())
-
-    override val context: CoroutineContext = scope.coroutineContext
-
-    /**
-     * The [TimerScheduler] to use for scheduling the interrupts.
-     */
-    private val scheduler = SimResourceSchedulerTrampoline(this.context, clock)
-
+) : SimAbstractMachine(platform) {
     override val cpus: List<SimProcessingUnit> = model.cpus.map { ProcessingUnitImpl(it) }
 
     /**
@@ -75,6 +61,8 @@ public class SimBareMetalMachine(
         scalingGovernor.createLogic(this.scalingDriver.createContext(cpu))
     }
 
+    override val parent: SimResourceSystem? = null
+
     init {
         scalingGovernors.forEach { it.onStart() }
     }
@@ -92,12 +80,6 @@ public class SimBareMetalMachine(
         powerDraw = scalingDriver.computePower()
     }
 
-    override fun close() {
-        super.close()
-
-        scope.cancel()
-    }
-
     /**
      * The [SimProcessingUnit] of this machine.
      */
@@ -105,7 +87,7 @@ public class SimBareMetalMachine(
         /**
          * The actual resource supporting the processing unit.
          */
-        private val source = SimResourceSource(model.frequency, scheduler)
+        private val source = SimResourceSource(model.frequency, interpreter, this@SimBareMetalMachine)
 
         override val speed: Double
             get() = source.speed
@@ -126,7 +108,7 @@ public class SimBareMetalMachine(
         }
 
         override fun close() {
-            source.interrupt()
+            source.close()
         }
     }
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
index 11aec2de..33e7e637 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
@@ -23,7 +23,9 @@
 package org.opendc.simulator.compute
 
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.resources.*
+import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.resources.SimResourceSwitch
+import org.opendc.simulator.resources.SimResourceSwitchMaxMin
 
 /**
  * A [SimHypervisor] that distributes the computing requirements of multiple [SimWorkload] on a single
@@ -31,13 +33,13 @@ import org.opendc.simulator.resources.*
  *
  * @param listener The hypervisor listener to use.
  */
-public class SimFairShareHypervisor(private val scheduler: SimResourceScheduler, private val listener: SimHypervisor.Listener? = null) : SimAbstractHypervisor() {
+public class SimFairShareHypervisor(private val interpreter: SimResourceInterpreter, private val listener: SimHypervisor.Listener? = null) : SimAbstractHypervisor(interpreter) {
 
     override fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean = true
 
     override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
         return SimResourceSwitchMaxMin(
-            scheduler,
+            interpreter, null,
             object : SimResourceSwitchMaxMin.Listener {
                 override fun onSliceFinish(
                     switch: SimResourceSwitchMaxMin,
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisorProvider.kt
index 2ab3ea09..68858cc1 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisorProvider.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisorProvider.kt
@@ -22,9 +22,7 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.resources.SimResourceSchedulerTrampoline
-import java.time.Clock
-import kotlin.coroutines.CoroutineContext
+import org.opendc.simulator.resources.SimResourceInterpreter
 
 /**
  * A [SimHypervisorProvider] for the [SimFairShareHypervisor] implementation.
@@ -32,7 +30,7 @@ import kotlin.coroutines.CoroutineContext
 public class SimFairShareHypervisorProvider : SimHypervisorProvider {
     override val id: String = "fair-share"
 
-    override fun create(context: CoroutineContext, clock: Clock, listener: SimHypervisor.Listener?): SimHypervisor {
-        return SimFairShareHypervisor(SimResourceSchedulerTrampoline(context, clock), listener)
+    override fun create(interpreter: SimResourceInterpreter, listener: SimHypervisor.Listener?): SimHypervisor {
+        return SimFairShareHypervisor(interpreter, listener)
     }
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisorProvider.kt
index b66020f4..d0753084 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisorProvider.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisorProvider.kt
@@ -22,8 +22,7 @@
 
 package org.opendc.simulator.compute
 
-import java.time.Clock
-import kotlin.coroutines.CoroutineContext
+import org.opendc.simulator.resources.SimResourceInterpreter
 
 /**
  * A service provider interface for constructing a [SimHypervisor].
@@ -40,5 +39,5 @@ public interface SimHypervisorProvider {
     /**
      * Create a [SimHypervisor] instance with the specified [listener].
      */
-    public fun create(context: CoroutineContext, clock: Clock, listener: SimHypervisor.Listener? = null): SimHypervisor
+    public fun create(interpreter: SimResourceInterpreter, listener: SimHypervisor.Listener? = null): SimHypervisor
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
index fd8e546f..afb47872 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
@@ -22,12 +22,14 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.resources.*
+import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.resources.SimResourceSwitch
+import org.opendc.simulator.resources.SimResourceSwitchExclusive
 
 /**
  * A [SimHypervisor] that allocates its sub-resources exclusively for the virtual machine that it hosts.
  */
-public class SimSpaceSharedHypervisor : SimAbstractHypervisor() {
+public class SimSpaceSharedHypervisor(interpreter: SimResourceInterpreter) : SimAbstractHypervisor(interpreter) {
     override fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean {
         return switch.inputs.size - switch.outputs.size >= model.cpus.size
     }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt
index 83b924d7..c017c8ab 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt
@@ -22,8 +22,7 @@
 
 package org.opendc.simulator.compute
 
-import java.time.Clock
-import kotlin.coroutines.CoroutineContext
+import org.opendc.simulator.resources.SimResourceInterpreter
 
 /**
  * A [SimHypervisorProvider] for the [SimSpaceSharedHypervisor] implementation.
@@ -31,7 +30,7 @@ import kotlin.coroutines.CoroutineContext
 public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider {
     override val id: String = "space-shared"
 
-    override fun create(context: CoroutineContext, clock: Clock, listener: SimHypervisor.Listener?): SimHypervisor {
-        return SimSpaceSharedHypervisor()
+    override fun create(interpreter: SimResourceInterpreter, listener: SimHypervisor.Listener?): SimHypervisor {
+        return SimSpaceSharedHypervisor(interpreter)
     }
 }