simulator: Introduce SimResourceScheduler

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

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

7 files changed, 133 insertions, 113 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 e78bcdac..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,12 +126,12 @@ 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)
 
@@ -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 8c15ec71..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,24 +34,26 @@ 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() {}
         }
 
-        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() {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
@@ -66,10 +68,11 @@ 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() {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
@@ -89,10 +92,11 @@ class SimResourceContextTest {
 
     @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() {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
@@ -101,4 +105,43 @@ class SimResourceContextTest {
         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 361a1516..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))
 
@@ -79,16 +77,15 @@ class SimResourceSourceTest {
             assertEquals(3000, clock.millis())
             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,9 +110,9 @@ 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 onNext(ctx: SimResourceContext): SimResourceCommand {
@@ -131,16 +127,15 @@ class SimResourceSourceTest {
         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 {
@@ -173,16 +168,15 @@ 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.onEvent(any(), eq(SimResourceEvent.Start)) }
@@ -193,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()) }
@@ -214,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()) }
@@ -238,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()) }
@@ -260,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()) }
@@ -283,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()) }
@@ -304,7 +293,6 @@ class SimResourceSourceTest {
 
             assertEquals(500, clock.millis())
         } finally {
-            scheduler.close()
             provider.close()
         }
     }
@@ -313,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()) }
@@ -325,7 +313,6 @@ class SimResourceSourceTest {
                 try {
                     provider.consume(consumer)
                 } finally {
-                    scheduler.close()
                     provider.close()
                 }
             }
@@ -334,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()) }
@@ -348,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 1b1f7790..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,7 +113,7 @@ 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 {
@@ -138,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)
 
@@ -159,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 e3ca5845..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)
@@ -124,8 +122,8 @@ internal class SimResourceTransformerTest {
     @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)
@@ -143,8 +141,8 @@ internal class SimResourceTransformerTest {
     @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)
@@ -162,8 +160,8 @@ internal class SimResourceTransformerTest {
     @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)
@@ -197,8 +195,8 @@ internal class SimResourceTransformerTest {
     @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)