simulator: Improve simulator resource model (#142)

This pull request improves the existing simulator resource model
that is at the core of all simulation models in OpenDC. 
Most importantly, we have changed the way of how metrics are reported by this layer.

* Add `SimResourceInterpreter` which is responsible for efficiently scheduling 
  communication between resources in OpenDC. The performance gain is in the 2x-5x range.
* Add uniform interface for exposing resource metrics (using `SimResourceCounters`).
* Split the CPUFreq subsystem in the compute simulator as it mixed responsibilities of 
  different layers.

**Breaking API Changes**

* Resource providers now accept a `SimResourceInterpreter` which is
   responsible for coordinating the communication between resources.
* `ScalingGovernor` is not part of the machine layer anymore. Instead, it should 
   move in the OS/Hypervisor layer.
* Workloads should now start CPU consumers using `cpu.startConsumer`.

7 files changed, 166 insertions, 144 deletions

diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
index 2b32300e..51024e80 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
@@ -33,6 +33,7 @@ import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.consumer.SimSpeedConsumerAdapter
 import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
  * Test suite for the [SimResourceAggregatorMaxMin] class.
@@ -41,7 +42,7 @@ import org.opendc.simulator.resources.consumer.SimWorkConsumer
 internal class SimResourceAggregatorMaxMinTest {
     @Test
     fun testSingleCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val forwarder = SimResourceForwarder()
@@ -58,7 +59,7 @@ internal class SimResourceAggregatorMaxMinTest {
         source.startConsumer(adapter)
 
         try {
-            aggregator.output.consume(consumer)
+            aggregator.consume(consumer)
             yield()
 
             assertAll(
@@ -66,13 +67,13 @@ internal class SimResourceAggregatorMaxMinTest {
                 { assertEquals(listOf(0.0, 0.5, 0.0), usage) }
             )
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
     @Test
     fun testDoubleCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
@@ -86,20 +87,20 @@ internal class SimResourceAggregatorMaxMinTest {
         val adapter = SimSpeedConsumerAdapter(consumer, usage::add)
 
         try {
-            aggregator.output.consume(adapter)
+            aggregator.consume(adapter)
             yield()
             assertAll(
                 { assertEquals(1000, clock.millis()) },
                 { assertEquals(listOf(0.0, 2.0, 0.0), usage) }
             )
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
     @Test
     fun testOvercommit() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
@@ -114,19 +115,19 @@ internal class SimResourceAggregatorMaxMinTest {
             .andThen(SimResourceCommand.Exit)
 
         try {
-            aggregator.output.consume(consumer)
+            aggregator.consume(consumer)
             yield()
             assertEquals(1000, clock.millis())
 
             verify(exactly = 2) { consumer.onNext(any()) }
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
     @Test
     fun testException() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
@@ -141,17 +142,17 @@ internal class SimResourceAggregatorMaxMinTest {
             .andThenThrows(IllegalStateException("Test Exception"))
 
         try {
-            assertThrows<IllegalStateException> { aggregator.output.consume(consumer) }
+            assertThrows<IllegalStateException> { aggregator.consume(consumer) }
             yield()
             assertEquals(SimResourceState.Pending, sources[0].state)
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
     @Test
     fun testAdjustCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
@@ -163,20 +164,20 @@ internal class SimResourceAggregatorMaxMinTest {
         val consumer = SimWorkConsumer(4.0, 1.0)
         try {
             coroutineScope {
-                launch { aggregator.output.consume(consumer) }
+                launch { aggregator.consume(consumer) }
                 delay(1000)
                 sources[0].capacity = 0.5
             }
             yield()
             assertEquals(2334, clock.millis())
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
     @Test
     fun testFailOverCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
@@ -188,14 +189,40 @@ internal class SimResourceAggregatorMaxMinTest {
         val consumer = SimWorkConsumer(1.0, 0.5)
         try {
             coroutineScope {
-                launch { aggregator.output.consume(consumer) }
+                launch { aggregator.consume(consumer) }
                 delay(500)
                 sources[0].capacity = 0.5
             }
             yield()
             assertEquals(1000, clock.millis())
         } finally {
-            aggregator.output.close()
+            aggregator.close()
+        }
+    }
+
+    @Test
+    fun testCounters() = runBlockingSimulation {
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
+        val sources = listOf(
+            SimResourceSource(1.0, scheduler),
+            SimResourceSource(1.0, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(4.0, 4.0, 1000))
+            .andThen(SimResourceCommand.Exit)
+
+        try {
+            aggregator.consume(consumer)
+            yield()
+            assertEquals(1000, clock.millis())
+            assertEquals(2.0, aggregator.counters.actual) { "Actual work mismatch" }
+        } finally {
+            aggregator.close()
         }
     }
 }
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
index 2e2d6588..6cb507ce 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
@@ -26,98 +26,109 @@ import io.mockk.*
 import kotlinx.coroutines.*
 import org.junit.jupiter.api.*
 import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.resources.impl.SimResourceContextImpl
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
- * A test suite for the [SimAbstractResourceContext] class.
+ * A test suite for the [SimResourceContextImpl] class.
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 class SimResourceContextTest {
     @Test
     fun testFlushWithoutCommand() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val interpreter = SimResourceInterpreterImpl(coroutineContext, clock)
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-        val context = object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
-            override fun onIdle(deadline: Long) {}
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
-            override fun onFinish() {}
+        val logic = object : SimResourceProviderLogic {
+            override fun onIdle(ctx: SimResourceControllableContext, deadline: Long): Long = deadline
+            override fun onConsume(ctx: SimResourceControllableContext, work: Double, limit: Double, deadline: Long): Long = deadline
+            override fun onFinish(ctx: SimResourceControllableContext) {}
         }
+        val context = SimResourceContextImpl(null, interpreter, consumer, logic)
 
-        context.flush(isIntermediate = false)
+        context.doUpdate(interpreter.clock.millis())
     }
 
     @Test
     fun testIntermediateFlush() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val interpreter = SimResourceInterpreterImpl(coroutineContext, clock)
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-        val context = spyk(object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
-            override fun onIdle(deadline: Long) {}
-            override fun onFinish() {}
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+        val logic = spyk(object : SimResourceProviderLogic {
+            override fun onIdle(ctx: SimResourceControllableContext, deadline: Long): Long = deadline
+            override fun onFinish(ctx: SimResourceControllableContext) {}
+            override fun onConsume(ctx: SimResourceControllableContext, work: Double, limit: Double, deadline: Long): Long = deadline
         })
+        val context = spyk(SimResourceContextImpl(null, interpreter, consumer, logic))
 
         context.start()
         delay(1) // Delay 1 ms to prevent hitting the fast path
-        context.flush(isIntermediate = true)
+        context.doUpdate(interpreter.clock.millis())
 
-        verify(exactly = 2) { context.onConsume(any(), any(), any()) }
+        verify(exactly = 2) { logic.onConsume(any(), any(), any(), any()) }
     }
 
     @Test
     fun testIntermediateFlushIdle() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val interpreter = SimResourceInterpreterImpl(coroutineContext, clock)
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10) andThen SimResourceCommand.Exit
 
-        val context = spyk(object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
-            override fun onIdle(deadline: Long) {}
-            override fun onFinish() {}
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+        val logic = spyk(object : SimResourceProviderLogic {
+            override fun onIdle(ctx: SimResourceControllableContext, deadline: Long): Long = deadline
+            override fun onFinish(ctx: SimResourceControllableContext) {}
+            override fun onConsume(ctx: SimResourceControllableContext, work: Double, limit: Double, deadline: Long): Long = deadline
         })
+        val context = spyk(SimResourceContextImpl(null, interpreter, consumer, logic))
 
         context.start()
         delay(5)
-        context.flush(isIntermediate = true)
+        context.invalidate()
         delay(5)
-        context.flush(isIntermediate = true)
+        context.invalidate()
 
         assertAll(
-            { verify(exactly = 2) { context.onIdle(any()) } },
-            { verify(exactly = 1) { context.onFinish() } }
+            { verify(exactly = 2) { logic.onIdle(any(), any()) } },
+            { verify(exactly = 1) { logic.onFinish(any()) } }
         )
     }
 
     @Test
     fun testDoubleStart() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val interpreter = SimResourceInterpreterImpl(coroutineContext, clock)
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10) andThen SimResourceCommand.Exit
 
-        val context = object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
-            override fun onIdle(deadline: Long) {}
-            override fun onFinish() {}
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+        val logic = object : SimResourceProviderLogic {
+            override fun onIdle(ctx: SimResourceControllableContext, deadline: Long): Long = deadline
+            override fun onFinish(ctx: SimResourceControllableContext) {}
+            override fun onConsume(ctx: SimResourceControllableContext, work: Double, limit: Double, deadline: Long): Long = deadline
         }
+        val context = SimResourceContextImpl(null, interpreter, consumer, logic)
 
         context.start()
-        assertThrows<IllegalStateException> { context.start() }
+
+        assertThrows<IllegalStateException> {
+            context.start()
+        }
     }
 
     @Test
     fun testIdempodentCapacityChange() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val interpreter = SimResourceInterpreterImpl(coroutineContext, clock)
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-        val context = object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
-            override fun onIdle(deadline: Long) {}
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
-            override fun onFinish() {}
+        val logic = object : SimResourceProviderLogic {
+            override fun onIdle(ctx: SimResourceControllableContext, deadline: Long): Long = deadline
+            override fun onFinish(ctx: SimResourceControllableContext) {}
+            override fun onConsume(ctx: SimResourceControllableContext, work: Double, limit: Double, deadline: Long): Long = deadline
         }
 
+        val context = SimResourceContextImpl(null, interpreter, consumer, logic)
+        context.capacity = 4200.0
         context.start()
         context.capacity = 4200.0
 
@@ -126,17 +137,19 @@ class SimResourceContextTest {
 
     @Test
     fun testFailureNoInfiniteLoop() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val interpreter = SimResourceInterpreterImpl(coroutineContext, clock)
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Exit
         every { consumer.onEvent(any(), SimResourceEvent.Exit) } throws IllegalStateException("onEvent")
         every { consumer.onFailure(any(), any()) } throws IllegalStateException("onFailure")
 
-        val context = object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
-            override fun onIdle(deadline: Long) {}
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
-            override fun onFinish() {}
-        }
+        val logic = spyk(object : SimResourceProviderLogic {
+            override fun onIdle(ctx: SimResourceControllableContext, deadline: Long): Long = deadline
+            override fun onFinish(ctx: SimResourceControllableContext) {}
+            override fun onConsume(ctx: SimResourceControllableContext, work: Double, limit: Double, deadline: Long): Long = deadline
+        })
+
+        val context = SimResourceContextImpl(null, interpreter, consumer, logic)
 
         context.start()
 
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
index 5e86088d..08d88093 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
@@ -32,6 +32,7 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.consumer.SimSpeedConsumerAdapter
 import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
  * A test suite for the [SimResourceSource] class.
@@ -40,7 +41,7 @@ import org.opendc.simulator.resources.consumer.SimWorkConsumer
 class SimResourceSourceTest {
     @Test
     fun testSpeed() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -63,7 +64,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testAdjustCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val provider = SimResourceSource(1.0, scheduler)
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
@@ -83,7 +84,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testSpeedLimit() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -110,7 +111,7 @@ class SimResourceSourceTest {
      */
     @Test
     fun testIntermediateInterrupt() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -133,7 +134,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testInterrupt() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
         lateinit var resCtx: SimResourceContext
@@ -174,7 +175,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testFailure() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -193,7 +194,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testExceptionPropagationOnNext() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -213,7 +214,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testConcurrentConsumption() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -236,7 +237,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testClosedConsumption() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -257,7 +258,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testCloseDuringConsumption() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -279,7 +280,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testIdle() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
@@ -301,7 +302,7 @@ class SimResourceSourceTest {
     fun testInfiniteSleep() {
         assertThrows<IllegalStateException> {
             runBlockingSimulation {
-                val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+                val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
                 val capacity = 4200.0
                 val provider = SimResourceSource(capacity, scheduler)
 
@@ -321,7 +322,7 @@ class SimResourceSourceTest {
 
     @Test
     fun testIncorrectDeadline() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
 
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
index 32b6d8ad..ad8d82e3 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
@@ -33,6 +33,7 @@ import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.consumer.SimSpeedConsumerAdapter
 import org.opendc.simulator.resources.consumer.SimTraceConsumer
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
  * Test suite for the [SimResourceSwitchExclusive] class.
@@ -44,7 +45,7 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testTrace() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val speed = mutableListOf<Double>()
 
@@ -66,7 +67,7 @@ internal class SimResourceSwitchExclusiveTest {
         source.startConsumer(adapter)
         switch.addInput(forwarder)
 
-        val provider = switch.addOutput(3200.0)
+        val provider = switch.newOutput()
 
         try {
             provider.consume(workload)
@@ -86,7 +87,7 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testRuntimeWorkload() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
         val workload = mockk<SimResourceConsumer>(relaxUnitFun = true)
@@ -97,7 +98,7 @@ internal class SimResourceSwitchExclusiveTest {
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(3200.0)
+        val provider = switch.newOutput()
 
         try {
             provider.consume(workload)
@@ -113,7 +114,7 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testTwoWorkloads() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
         val workload = object : SimResourceConsumer {
@@ -141,7 +142,7 @@ internal class SimResourceSwitchExclusiveTest {
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(3200.0)
+        val provider = switch.newOutput()
 
         try {
             provider.consume(workload)
@@ -158,7 +159,7 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testConcurrentWorkloadFails() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
         val workload = mockk<SimResourceConsumer>(relaxUnitFun = true)
@@ -169,7 +170,7 @@ internal class SimResourceSwitchExclusiveTest {
 
         switch.addInput(source)
 
-        switch.addOutput(3200.0)
-        assertThrows<IllegalStateException> { switch.addOutput(3200.0) }
+        switch.newOutput()
+        assertThrows<IllegalStateException> { switch.newOutput() }
     }
 }
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
index e7dec172..e4292ec0 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
@@ -32,6 +32,7 @@ import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.consumer.SimTraceConsumer
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
  * Test suite for the [SimResourceSwitch] implementations
@@ -40,13 +41,13 @@ import org.opendc.simulator.resources.consumer.SimTraceConsumer
 internal class SimResourceSwitchMaxMinTest {
     @Test
     fun testSmoke() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val switch = SimResourceSwitchMaxMin(scheduler)
 
         val sources = List(2) { SimResourceSource(2000.0, scheduler) }
         sources.forEach { switch.addInput(it) }
 
-        val provider = switch.addOutput(1000.0)
+        val provider = switch.newOutput()
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(1.0, 1.0) andThen SimResourceCommand.Exit
@@ -64,27 +65,7 @@ internal class SimResourceSwitchMaxMinTest {
      */
     @Test
     fun testOvercommittedSingle() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
-
-        val listener = object : SimResourceSwitchMaxMin.Listener {
-            var totalRequestedWork = 0L
-            var totalGrantedWork = 0L
-            var totalOvercommittedWork = 0L
-
-            override fun onSliceFinish(
-                switch: SimResourceSwitchMaxMin,
-                requestedWork: Long,
-                grantedWork: Long,
-                overcommittedWork: Long,
-                interferedWork: Long,
-                cpuUsage: Double,
-                cpuDemand: Double
-            ) {
-                totalRequestedWork += requestedWork
-                totalGrantedWork += grantedWork
-                totalOvercommittedWork += overcommittedWork
-            }
-        }
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val duration = 5 * 60L
         val workload =
@@ -97,8 +78,8 @@ internal class SimResourceSwitchMaxMinTest {
                 ),
             )
 
-        val switch = SimResourceSwitchMaxMin(scheduler, listener)
-        val provider = switch.addOutput(3200.0)
+        val switch = SimResourceSwitchMaxMin(scheduler)
+        val provider = switch.newOutput()
 
         try {
             switch.addInput(SimResourceSource(3200.0, scheduler))
@@ -109,9 +90,9 @@ internal class SimResourceSwitchMaxMinTest {
         }
 
         assertAll(
-            { assertEquals(1113300, listener.totalRequestedWork, "Requested Burst does not match") },
-            { assertEquals(1023300, listener.totalGrantedWork, "Granted Burst does not match") },
-            { assertEquals(90000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(1113300.0, switch.counters.demand, "Requested work does not match") },
+            { assertEquals(1023300.0, switch.counters.actual, "Actual work does not match") },
+            { assertEquals(90000.0, switch.counters.overcommit, "Overcommitted work does not match") },
             { assertEquals(1200000, clock.millis()) }
         )
     }
@@ -121,27 +102,7 @@ internal class SimResourceSwitchMaxMinTest {
      */
     @Test
     fun testOvercommittedDual() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
-
-        val listener = object : SimResourceSwitchMaxMin.Listener {
-            var totalRequestedWork = 0L
-            var totalGrantedWork = 0L
-            var totalOvercommittedWork = 0L
-
-            override fun onSliceFinish(
-                switch: SimResourceSwitchMaxMin,
-                requestedWork: Long,
-                grantedWork: Long,
-                overcommittedWork: Long,
-                interferedWork: Long,
-                cpuUsage: Double,
-                cpuDemand: Double
-            ) {
-                totalRequestedWork += requestedWork
-                totalGrantedWork += grantedWork
-                totalOvercommittedWork += overcommittedWork
-            }
-        }
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
         val duration = 5 * 60L
         val workloadA =
@@ -163,9 +124,9 @@ internal class SimResourceSwitchMaxMinTest {
                 )
             )
 
-        val switch = SimResourceSwitchMaxMin(scheduler, listener)
-        val providerA = switch.addOutput(3200.0)
-        val providerB = switch.addOutput(3200.0)
+        val switch = SimResourceSwitchMaxMin(scheduler)
+        val providerA = switch.newOutput()
+        val providerB = switch.newOutput()
 
         try {
             switch.addInput(SimResourceSource(3200.0, scheduler))
@@ -180,9 +141,9 @@ internal class SimResourceSwitchMaxMinTest {
             switch.close()
         }
         assertAll(
-            { assertEquals(2082000, listener.totalRequestedWork, "Requested Burst does not match") },
-            { assertEquals(1062000, listener.totalGrantedWork, "Granted Burst does not match") },
-            { assertEquals(1020000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(2073600.0, switch.counters.demand, "Requested work does not match") },
+            { assertEquals(1053600.0, switch.counters.actual, "Granted work does not match") },
+            { assertEquals(1020000.0, switch.counters.overcommit, "Overcommitted work does not match") },
             { assertEquals(1200000, clock.millis()) }
         )
     }
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt
index 880e1755..810052b8 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt
@@ -32,6 +32,7 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
  * A test suite for the [SimResourceTransformer] class.
@@ -41,7 +42,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testExitImmediately() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
         launch {
@@ -61,7 +62,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testExit() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
         launch {
@@ -122,7 +123,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testCancelStartedDelegate() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
@@ -141,7 +142,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testCancelPropagation() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
@@ -160,7 +161,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testExitPropagation() = runBlockingSimulation {
         val forwarder = SimResourceForwarder(isCoupled = true)
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
@@ -176,7 +177,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testAdjustCapacity() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(1.0, scheduler)
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
@@ -195,7 +196,7 @@ internal class SimResourceTransformerTest {
     @Test
     fun testTransformExit() = runBlockingSimulation {
         val forwarder = SimResourceTransformer { _, _ -> SimResourceCommand.Exit }
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(1.0, scheduler)
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
@@ -205,4 +206,21 @@ internal class SimResourceTransformerTest {
         assertEquals(0, clock.millis())
         verify(exactly = 1) { consumer.onNext(any()) }
     }
+
+    @Test
+    fun testCounters() = runBlockingSimulation {
+        val forwarder = SimResourceForwarder()
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
+        val source = SimResourceSource(1.0, scheduler)
+
+        val consumer = SimWorkConsumer(2.0, 1.0)
+        source.startConsumer(forwarder)
+
+        forwarder.consume(consumer)
+
+        assertEquals(source.counters.actual, forwarder.counters.actual) { "Actual work" }
+        assertEquals(source.counters.demand, forwarder.counters.demand) { "Work demand" }
+        assertEquals(source.counters.overcommit, forwarder.counters.overcommit) { "Overcommitted work" }
+        assertEquals(2000, clock.millis())
+    }
 }
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt
index ac8b5814..db4fe856 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 
 /**
  * A test suite for the [SimWorkConsumer] class.
@@ -35,7 +36,7 @@ import org.opendc.simulator.resources.consumer.SimWorkConsumer
 internal class SimWorkConsumerTest {
     @Test
     fun testSmoke() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val provider = SimResourceSource(1.0, scheduler)
 
         val consumer = SimWorkConsumer(1.0, 1.0)
@@ -50,7 +51,7 @@ internal class SimWorkConsumerTest {
 
     @Test
     fun testUtilization() = runBlockingSimulation {
-        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val provider = SimResourceSource(1.0, scheduler)
 
         val consumer = SimWorkConsumer(1.0, 0.5)