simulator: Add support for central resource scheduling (#126)

This pull request adds support for central resource scheduling. 
This enables possible optimizations in the future where we can efficiently 
schedule resource updates.

* Introduce `SimResourceScheduler` which centralizes the logic for 
  scheduling resource interrupts.
* Fix benchmarks
* Add generic approach for reporting resource events
* Simplify scheduling logic of resource aggregator.

**Breaking API Changes**

* Classes in `opendc-simulator-resources` now take `SimResourceScheduler` as 
   opposed to a `CoroutineContext` and `Clock`.

7 files changed, 162 insertions, 135 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 e272abb8..2b32300e 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,7 +33,6 @@ 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.utils.TimerScheduler
 
 /**
  * Test suite for the [SimResourceAggregatorMaxMin] class.
@@ -42,19 +41,19 @@ import org.opendc.utils.TimerScheduler
 internal class SimResourceAggregatorMaxMinTest {
     @Test
     fun testSingleCapacity() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
-        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val forwarder = SimResourceForwarder()
         val sources = listOf(
             forwarder,
-            SimResourceSource(1.0, clock, scheduler)
+            SimResourceSource(1.0, scheduler)
         )
         sources.forEach(aggregator::addInput)
 
         val consumer = SimWorkConsumer(1.0, 0.5)
         val usage = mutableListOf<Double>()
-        val source = SimResourceSource(1.0, clock, scheduler)
+        val source = SimResourceSource(1.0, scheduler)
         val adapter = SimSpeedConsumerAdapter(forwarder, usage::add)
         source.startConsumer(adapter)
 
@@ -73,12 +72,12 @@ internal class SimResourceAggregatorMaxMinTest {
 
     @Test
     fun testDoubleCapacity() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
-        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
-            SimResourceSource(1.0, clock, scheduler),
-            SimResourceSource(1.0, clock, scheduler)
+            SimResourceSource(1.0, scheduler),
+            SimResourceSource(1.0, scheduler)
         )
         sources.forEach(aggregator::addInput)
 
@@ -100,12 +99,12 @@ internal class SimResourceAggregatorMaxMinTest {
 
     @Test
     fun testOvercommit() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
-        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
-            SimResourceSource(1.0, clock, scheduler),
-            SimResourceSource(1.0, clock, scheduler)
+            SimResourceSource(1.0, scheduler),
+            SimResourceSource(1.0, scheduler)
         )
         sources.forEach(aggregator::addInput)
 
@@ -127,19 +126,19 @@ internal class SimResourceAggregatorMaxMinTest {
 
     @Test
     fun testException() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
-        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
-            SimResourceSource(1.0, clock, scheduler),
-            SimResourceSource(1.0, clock, scheduler)
+            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(1.0, 1.0))
-            .andThenThrows(IllegalStateException())
+            .andThenThrows(IllegalStateException("Test Exception"))
 
         try {
             assertThrows<IllegalStateException> { aggregator.output.consume(consumer) }
@@ -152,12 +151,12 @@ internal class SimResourceAggregatorMaxMinTest {
 
     @Test
     fun testAdjustCapacity() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
-        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
-            SimResourceSource(1.0, clock, scheduler),
-            SimResourceSource(1.0, clock, scheduler)
+            SimResourceSource(1.0, scheduler),
+            SimResourceSource(1.0, scheduler)
         )
         sources.forEach(aggregator::addInput)
 
@@ -177,12 +176,12 @@ internal class SimResourceAggregatorMaxMinTest {
 
     @Test
     fun testFailOverCapacity() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
-        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val aggregator = SimResourceAggregatorMaxMin(scheduler)
         val sources = listOf(
-            SimResourceSource(1.0, clock, scheduler),
-            SimResourceSource(1.0, clock, scheduler)
+            SimResourceSource(1.0, scheduler),
+            SimResourceSource(1.0, scheduler)
         )
         sources.forEach(aggregator::addInput)
 
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 be909556..2e2d6588 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
@@ -34,26 +34,28 @@ import org.opendc.simulator.core.runBlockingSimulation
 class SimResourceContextTest {
     @Test
     fun testFlushWithoutCommand() = runBlockingSimulation {
+        val scheduler = SimResourceSchedulerTrampoline(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, clock, consumer) {
+        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(cause: Throwable?) {}
+            override fun onFinish() {}
         }
 
-        context.flush()
+        context.flush(isIntermediate = false)
     }
 
     @Test
     fun testIntermediateFlush() = runBlockingSimulation {
+        val scheduler = SimResourceSchedulerTrampoline(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, clock, consumer) {
+        val context = spyk(object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
             override fun onIdle(deadline: Long) {}
-            override fun onFinish(cause: Throwable?) {}
+            override fun onFinish() {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
         })
 
@@ -66,12 +68,13 @@ class SimResourceContextTest {
 
     @Test
     fun testIntermediateFlushIdle() = runBlockingSimulation {
+        val scheduler = SimResourceSchedulerTrampoline(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, clock, consumer) {
+        val context = spyk(object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
             override fun onIdle(deadline: Long) {}
-            override fun onFinish(cause: Throwable?) {}
+            override fun onFinish() {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
         })
 
@@ -83,22 +86,62 @@ class SimResourceContextTest {
 
         assertAll(
             { verify(exactly = 2) { context.onIdle(any()) } },
-            { verify(exactly = 1) { context.onFinish(null) } }
+            { verify(exactly = 1) { context.onFinish() } }
         )
     }
 
     @Test
     fun testDoubleStart() = runBlockingSimulation {
+        val scheduler = SimResourceSchedulerTrampoline(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, clock, consumer) {
+        val context = object : SimAbstractResourceContext(4200.0, scheduler, consumer) {
             override fun onIdle(deadline: Long) {}
-            override fun onFinish(cause: Throwable?) {}
+            override fun onFinish() {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
         }
 
         context.start()
         assertThrows<IllegalStateException> { context.start() }
     }
+
+    @Test
+    fun testIdempodentCapacityChange() = runBlockingSimulation {
+        val scheduler = SimResourceSchedulerTrampoline(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() {}
+        }
+
+        context.start()
+        context.capacity = 4200.0
+
+        verify(exactly = 0) { consumer.onEvent(any(), SimResourceEvent.Capacity) }
+    }
+
+    @Test
+    fun testFailureNoInfiniteLoop() = runBlockingSimulation {
+        val scheduler = SimResourceSchedulerTrampoline(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() {}
+        }
+
+        context.start()
+
+        delay(1)
+
+        verify(exactly = 1) { consumer.onFailure(any(), any()) }
+    }
 }
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 39f74481..5e86088d 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,7 +32,6 @@ 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.utils.TimerScheduler
 
 /**
  * A test suite for the [SimResourceSource] class.
@@ -41,9 +40,9 @@ import org.opendc.utils.TimerScheduler
 class SimResourceSourceTest {
     @Test
     fun testSpeed() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -58,15 +57,14 @@ class SimResourceSourceTest {
 
             assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testAdjustCapacity() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(1.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val provider = SimResourceSource(1.0, scheduler)
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
 
@@ -77,18 +75,17 @@ class SimResourceSourceTest {
                 provider.capacity = 0.5
             }
             assertEquals(3000, clock.millis())
-            verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
+            verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Capacity) }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testSpeedLimit() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -103,7 +100,6 @@ class SimResourceSourceTest {
 
             assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
@@ -114,39 +110,42 @@ class SimResourceSourceTest {
      */
     @Test
     fun testIntermediateInterrupt() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = object : SimResourceConsumer {
-            override fun onStart(ctx: SimResourceContext) {
-                ctx.interrupt()
-            }
-
             override fun onNext(ctx: SimResourceContext): SimResourceCommand {
                 return SimResourceCommand.Exit
             }
+
+            override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
+                ctx.interrupt()
+            }
         }
 
         try {
             provider.consume(consumer)
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testInterrupt() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
         lateinit var resCtx: SimResourceContext
 
         val consumer = object : SimResourceConsumer {
             var isFirst = true
-            override fun onStart(ctx: SimResourceContext) {
-                resCtx = ctx
+
+            override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
+                when (event) {
+                    SimResourceEvent.Start -> resCtx = ctx
+                    else -> {}
+                }
             }
 
             override fun onNext(ctx: SimResourceContext): SimResourceCommand {
@@ -169,19 +168,18 @@ class SimResourceSourceTest {
 
             assertEquals(0, clock.millis())
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testFailure() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
-        every { consumer.onStart(any()) }
+        every { consumer.onEvent(any(), eq(SimResourceEvent.Start)) }
             .throws(IllegalStateException())
 
         try {
@@ -189,16 +187,15 @@ class SimResourceSourceTest {
                 provider.consume(consumer)
             }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testExceptionPropagationOnNext() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -210,16 +207,15 @@ class SimResourceSourceTest {
                 provider.consume(consumer)
             }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testConcurrentConsumption() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -234,16 +230,15 @@ class SimResourceSourceTest {
                 }
             }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testClosedConsumption() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -256,16 +251,15 @@ class SimResourceSourceTest {
                 provider.consume(consumer)
             }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testCloseDuringConsumption() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -279,16 +273,15 @@ class SimResourceSourceTest {
 
             assertEquals(500, clock.millis())
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
 
     @Test
     fun testIdle() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -300,7 +293,6 @@ class SimResourceSourceTest {
 
             assertEquals(500, clock.millis())
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
@@ -309,9 +301,9 @@ class SimResourceSourceTest {
     fun testInfiniteSleep() {
         assertThrows<IllegalStateException> {
             runBlockingSimulation {
-                val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+                val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
                 val capacity = 4200.0
-                val provider = SimResourceSource(capacity, clock, scheduler)
+                val provider = SimResourceSource(capacity, scheduler)
 
                 val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
                 every { consumer.onNext(any()) }
@@ -321,7 +313,6 @@ class SimResourceSourceTest {
                 try {
                     provider.consume(consumer)
                 } finally {
-                    scheduler.close()
                     provider.close()
                 }
             }
@@ -330,9 +321,9 @@ class SimResourceSourceTest {
 
     @Test
     fun testIncorrectDeadline() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
         val capacity = 4200.0
-        val provider = SimResourceSource(capacity, clock, scheduler)
+        val provider = SimResourceSource(capacity, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) }
@@ -344,7 +335,6 @@ class SimResourceSourceTest {
 
             assertThrows<IllegalArgumentException> { provider.consume(consumer) }
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
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 f7d17867..32b6d8ad 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,7 +33,6 @@ 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.utils.TimerScheduler
 
 /**
  * Test suite for the [SimResourceSwitchExclusive] class.
@@ -45,7 +44,7 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testTrace() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
         val speed = mutableListOf<Double>()
 
@@ -61,7 +60,7 @@ internal class SimResourceSwitchExclusiveTest {
             )
 
         val switch = SimResourceSwitchExclusive()
-        val source = SimResourceSource(3200.0, clock, scheduler)
+        val source = SimResourceSource(3200.0, scheduler)
         val forwarder = SimResourceForwarder()
         val adapter = SimSpeedConsumerAdapter(forwarder, speed::add)
         source.startConsumer(adapter)
@@ -87,14 +86,14 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testRuntimeWorkload() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
         val workload = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { workload.onNext(any()) } returns SimResourceCommand.Consume(duration / 1000.0, 1.0) andThen SimResourceCommand.Exit
 
         val switch = SimResourceSwitchExclusive()
-        val source = SimResourceSource(3200.0, clock, scheduler)
+        val source = SimResourceSource(3200.0, scheduler)
 
         switch.addInput(source)
 
@@ -114,14 +113,17 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testTwoWorkloads() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
         val workload = object : SimResourceConsumer {
             var isFirst = true
 
-            override fun onStart(ctx: SimResourceContext) {
-                isFirst = true
+            override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
+                when (event) {
+                    SimResourceEvent.Start -> isFirst = true
+                    else -> {}
+                }
             }
 
             override fun onNext(ctx: SimResourceContext): SimResourceCommand {
@@ -135,7 +137,7 @@ internal class SimResourceSwitchExclusiveTest {
         }
 
         val switch = SimResourceSwitchExclusive()
-        val source = SimResourceSource(3200.0, clock, scheduler)
+        val source = SimResourceSource(3200.0, scheduler)
 
         switch.addInput(source)
 
@@ -156,14 +158,14 @@ internal class SimResourceSwitchExclusiveTest {
      */
     @Test
     fun testConcurrentWorkloadFails() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
         val workload = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { workload.onNext(any()) } returns SimResourceCommand.Consume(duration / 1000.0, 1.0) andThen SimResourceCommand.Exit
 
         val switch = SimResourceSwitchExclusive()
-        val source = SimResourceSource(3200.0, clock, scheduler)
+        val source = SimResourceSource(3200.0, scheduler)
 
         switch.addInput(source)
 
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 7416f277..e7dec172 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,7 +32,6 @@ 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.utils.TimerScheduler
 
 /**
  * Test suite for the [SimResourceSwitch] implementations
@@ -41,10 +40,10 @@ import org.opendc.utils.TimerScheduler
 internal class SimResourceSwitchMaxMinTest {
     @Test
     fun testSmoke() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val switch = SimResourceSwitchMaxMin(clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val switch = SimResourceSwitchMaxMin(scheduler)
 
-        val sources = List(2) { SimResourceSource(2000.0, clock, scheduler) }
+        val sources = List(2) { SimResourceSource(2000.0, scheduler) }
         sources.forEach { switch.addInput(it) }
 
         val provider = switch.addOutput(1000.0)
@@ -57,7 +56,6 @@ internal class SimResourceSwitchMaxMinTest {
             yield()
         } finally {
             switch.close()
-            scheduler.close()
         }
     }
 
@@ -66,7 +64,7 @@ internal class SimResourceSwitchMaxMinTest {
      */
     @Test
     fun testOvercommittedSingle() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
         val listener = object : SimResourceSwitchMaxMin.Listener {
             var totalRequestedWork = 0L
@@ -99,16 +97,15 @@ internal class SimResourceSwitchMaxMinTest {
                 ),
             )
 
-        val switch = SimResourceSwitchMaxMin(clock, listener)
+        val switch = SimResourceSwitchMaxMin(scheduler, listener)
         val provider = switch.addOutput(3200.0)
 
         try {
-            switch.addInput(SimResourceSource(3200.0, clock, scheduler))
+            switch.addInput(SimResourceSource(3200.0, scheduler))
             provider.consume(workload)
             yield()
         } finally {
             switch.close()
-            scheduler.close()
         }
 
         assertAll(
@@ -124,7 +121,7 @@ internal class SimResourceSwitchMaxMinTest {
      */
     @Test
     fun testOvercommittedDual() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
 
         val listener = object : SimResourceSwitchMaxMin.Listener {
             var totalRequestedWork = 0L
@@ -166,12 +163,12 @@ internal class SimResourceSwitchMaxMinTest {
                 )
             )
 
-        val switch = SimResourceSwitchMaxMin(clock, listener)
+        val switch = SimResourceSwitchMaxMin(scheduler, listener)
         val providerA = switch.addOutput(3200.0)
         val providerB = switch.addOutput(3200.0)
 
         try {
-            switch.addInput(SimResourceSource(3200.0, clock, scheduler))
+            switch.addInput(SimResourceSource(3200.0, scheduler))
 
             coroutineScope {
                 launch { providerA.consume(workloadA) }
@@ -181,7 +178,6 @@ internal class SimResourceSwitchMaxMinTest {
             yield()
         } finally {
             switch.close()
-            scheduler.close()
         }
         assertAll(
             { assertEquals(2082000, listener.totalRequestedWork, "Requested Burst does not match") },
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 d2ad73bc..880e1755 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,7 +32,6 @@ 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.utils.TimerScheduler
 
 /**
  * A test suite for the [SimResourceTransformer] class.
@@ -42,8 +41,8 @@ internal class SimResourceTransformerTest {
     @Test
     fun testExitImmediately() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(2000.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, scheduler)
 
         launch {
             source.consume(forwarder)
@@ -57,14 +56,13 @@ internal class SimResourceTransformerTest {
         })
 
         forwarder.close()
-        scheduler.close()
     }
 
     @Test
     fun testExit() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(2000.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, scheduler)
 
         launch {
             source.consume(forwarder)
@@ -118,14 +116,14 @@ internal class SimResourceTransformerTest {
         forwarder.startConsumer(consumer)
         forwarder.cancel()
 
-        verify(exactly = 0) { consumer.onFinish(any(), null) }
+        verify(exactly = 0) { consumer.onEvent(any(), SimResourceEvent.Exit) }
     }
 
     @Test
     fun testCancelStartedDelegate() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(2000.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10)
@@ -136,15 +134,15 @@ internal class SimResourceTransformerTest {
         yield()
         forwarder.cancel()
 
-        verify(exactly = 1) { consumer.onStart(any()) }
-        verify(exactly = 1) { consumer.onFinish(any(), null) }
+        verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Start) }
+        verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Exit) }
     }
 
     @Test
     fun testCancelPropagation() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(2000.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10)
@@ -155,15 +153,15 @@ internal class SimResourceTransformerTest {
         yield()
         source.cancel()
 
-        verify(exactly = 1) { consumer.onStart(any()) }
-        verify(exactly = 1) { consumer.onFinish(any(), null) }
+        verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Start) }
+        verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Exit) }
     }
 
     @Test
     fun testExitPropagation() = runBlockingSimulation {
         val forwarder = SimResourceForwarder(isCoupled = true)
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(2000.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, scheduler)
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Exit
@@ -178,8 +176,8 @@ internal class SimResourceTransformerTest {
     @Test
     fun testAdjustCapacity() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(1.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(1.0, scheduler)
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
         source.startConsumer(forwarder)
@@ -191,14 +189,14 @@ internal class SimResourceTransformerTest {
         }
 
         assertEquals(3000, clock.millis())
-        verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
+        verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Capacity) }
     }
 
     @Test
     fun testTransformExit() = runBlockingSimulation {
         val forwarder = SimResourceTransformer { _, _ -> SimResourceCommand.Exit }
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(1.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val source = SimResourceSource(1.0, scheduler)
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
         source.startConsumer(forwarder)
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 bf58b1b6..ac8b5814 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,7 +27,6 @@ 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.utils.TimerScheduler
 
 /**
  * A test suite for the [SimWorkConsumer] class.
@@ -36,8 +35,8 @@ import org.opendc.utils.TimerScheduler
 internal class SimWorkConsumerTest {
     @Test
     fun testSmoke() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(1.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val provider = SimResourceSource(1.0, scheduler)
 
         val consumer = SimWorkConsumer(1.0, 1.0)
 
@@ -51,8 +50,8 @@ internal class SimWorkConsumerTest {
 
     @Test
     fun testUtilization() = runBlockingSimulation {
-        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(1.0, clock, scheduler)
+        val scheduler = SimResourceSchedulerTrampoline(coroutineContext, clock)
+        val provider = SimResourceSource(1.0, scheduler)
 
         val consumer = SimWorkConsumer(1.0, 0.5)