simulator: Add uniform resource consumption model (v2)

This is the second pull request in the series of pull requests to add a uniform resource consumption model to OpenDC. This pull request focusses on adding dynamic capacity negotiation and propagation between resource consumer and resource provider:

* The generic resource constraint is removed from the interfaces of `opendc-simulator-resources`. Users of the API are expected to use the untyped variants where only the capacity needs to be specified. Users are expected to build higher-level abstractions on top of these interface to represent actual resources (e.g., CPU, disk or network).
* Added benchmarks for the most important implementations of `opendc-simulator-resources`. This allows us to quantify the effects of changes on the runtime.
* The `SimResourceSwitchMaxMin` has been split into a `SimResourceAggregatorMaxMin` and `SimResourceDistributorMaxMin` which respectively aggregate input resources and distribute output resources using max-min fair sharing.
* The `SimResourceConsumer` interface has a new method for receiving capacity change events: `onCapacityChanged(ctx, isThrottled)`

**Breaking API Changes**

*  All interfaces in `opendc-simulator-resources`.

7 files changed, 608 insertions, 288 deletions

diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
new file mode 100644
index 00000000..de864c1c
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
@@ -0,0 +1,208 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.simulator.resources
+
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.test.runBlockingTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertAll
+import org.junit.jupiter.api.assertThrows
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+import org.opendc.utils.TimerScheduler
+
+/**
+ * Test suite for the [SimResourceAggregatorMaxMin] class.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class SimResourceAggregatorMaxMinTest {
+    @Test
+    fun testSingleCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = SimWorkConsumer(1.0, 0.5)
+        val usage = mutableListOf<Double>()
+        val job = launch { sources[0].speed.toList(usage) }
+
+        try {
+            aggregator.output.consume(consumer)
+            yield()
+
+            assertAll(
+                { assertEquals(1000, currentTime) },
+                { assertEquals(listOf(0.0, 0.5, 0.0), usage) }
+            )
+        } finally {
+            aggregator.output.close()
+            job.cancel()
+        }
+    }
+
+    @Test
+    fun testDoubleCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = SimWorkConsumer(2.0, 1.0)
+        val usage = mutableListOf<Double>()
+        val job = launch { sources[0].speed.toList(usage) }
+
+        try {
+            aggregator.output.consume(consumer)
+            yield()
+            assertAll(
+                { assertEquals(1000, currentTime) },
+                { assertEquals(listOf(0.0, 1.0, 0.0), usage) }
+            )
+        } finally {
+            aggregator.output.close()
+            job.cancel()
+        }
+    }
+
+    @Test
+    fun testOvercommit() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, 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.output.consume(consumer)
+            yield()
+            assertEquals(1000, currentTime)
+
+            verify(exactly = 2) { consumer.onNext(any()) }
+        } finally {
+            aggregator.output.close()
+        }
+    }
+
+    @Test
+    fun testException() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1.0, 1.0))
+            .andThenThrows(IllegalStateException())
+
+        try {
+            assertThrows<IllegalStateException> { aggregator.output.consume(consumer) }
+            yield()
+            assertEquals(SimResourceState.Pending, sources[0].state)
+        } finally {
+            aggregator.output.close()
+        }
+    }
+
+    @Test
+    fun testAdjustCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = SimWorkConsumer(4.0, 1.0)
+        try {
+            coroutineScope {
+                launch { aggregator.output.consume(consumer) }
+                delay(1000)
+                sources[0].capacity = 0.5
+            }
+            yield()
+            assertEquals(2334, currentTime)
+        } finally {
+            aggregator.output.close()
+        }
+    }
+
+    @Test
+    fun testFailOverCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = SimWorkConsumer(1.0, 0.5)
+        try {
+            coroutineScope {
+                launch { aggregator.output.consume(consumer) }
+                delay(500)
+                sources[0].capacity = 0.5
+            }
+            yield()
+            assertEquals(1000, currentTime)
+        } finally {
+            aggregator.output.close()
+        }
+    }
+}
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
index e7642dc1..030a0f6b 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
@@ -22,11 +22,10 @@
 
 package org.opendc.simulator.resources
 
+import io.mockk.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.*
-import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.Assertions.assertTrue
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 
 /**
@@ -34,39 +33,17 @@ import org.opendc.simulator.utils.DelayControllerClockAdapter
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 class SimResourceContextTest {
-    data class SimCpu(val speed: Double) : SimResource {
-        override val capacity: Double
-            get() = speed
-    }
-
     @Test
     fun testFlushWithoutCommand() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
 
-        val resource = SimCpu(4200.0)
-
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(10.0, 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
-
-        val context = object : SimAbstractResourceContext<SimCpu>(resource, clock, consumer) {
-            override fun onIdle(deadline: Long) {
-            }
-
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {
-            }
-
-            override fun onFinish() {
-            }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-            override fun onFailure(cause: Throwable) {
-            }
+        val context = object : SimAbstractResourceContext(4200.0, clock, consumer) {
+            override fun onIdle(deadline: Long) {}
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+            override fun onFinish(cause: Throwable?) {}
         }
 
         context.flush()
@@ -75,72 +52,35 @@ class SimResourceContextTest {
     @Test
     fun testIntermediateFlush() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
-        val resource = SimCpu(4200.0)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(10.0, 1.0)
-            }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
-
-        var counter = 0
-        val context = object : SimAbstractResourceContext<SimCpu>(resource, clock, consumer) {
-            override fun onIdle(deadline: Long) {
-            }
-
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {
-                counter++
-            }
-
-            override fun onFinish() {
-            }
-
-            override fun onFailure(cause: Throwable) {
-            }
-        }
+        val context = spyk(object : SimAbstractResourceContext(4200.0, clock, consumer) {
+            override fun onIdle(deadline: Long) {}
+            override fun onFinish(cause: Throwable?) {}
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+        })
 
         context.start()
         delay(1) // Delay 1 ms to prevent hitting the fast path
         context.flush(isIntermediate = true)
-        assertEquals(2, counter)
+
+        verify(exactly = 2) { context.onConsume(any(), any(), any()) }
     }
 
     @Test
     fun testIntermediateFlushIdle() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
-        val resource = SimCpu(4200.0)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Idle(10)
-            }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10) andThen SimResourceCommand.Exit
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
-
-        var counter = 0
-        var isFinished = false
-        val context = object : SimAbstractResourceContext<SimCpu>(resource, clock, consumer) {
-            override fun onIdle(deadline: Long) {
-                counter++
-            }
-
-            override fun onConsume(work: Double, limit: Double, deadline: Long) {
-            }
-
-            override fun onFinish() {
-                isFinished = true
-            }
-
-            override fun onFailure(cause: Throwable) {
-            }
-        }
+        val context = spyk(object : SimAbstractResourceContext(4200.0, clock, consumer) {
+            override fun onIdle(deadline: Long) {}
+            override fun onFinish(cause: Throwable?) {}
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+        })
 
         context.start()
         delay(5)
@@ -149,8 +89,25 @@ class SimResourceContextTest {
         context.flush(isIntermediate = true)
 
         assertAll(
-            { assertEquals(1, counter) },
-            { assertTrue(isFinished) }
+            { verify(exactly = 2) { context.onIdle(any()) } },
+            { verify(exactly = 1) { context.onFinish(null) } }
         )
     }
+
+    @Test
+    fun testDoubleStart() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+
+        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) {
+            override fun onIdle(deadline: Long) {}
+            override fun onFinish(cause: Throwable?) {}
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {}
+        }
+
+        context.start()
+        assertThrows<IllegalStateException> { context.start() }
+    }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
index ced1bd98..143dbca9 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
@@ -22,10 +22,16 @@
 
 package org.opendc.simulator.resources
 
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.launch
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.spyk
+import io.mockk.verify
+import kotlinx.coroutines.*
 import kotlinx.coroutines.test.runBlockingTest
+import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
 
@@ -34,59 +40,147 @@ import org.opendc.utils.TimerScheduler
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class SimResourceForwarderTest {
-
-    data class SimCpu(val speed: Double) : SimResource {
-        override val capacity: Double
-            get() = speed
-    }
-
     @Test
     fun testExitImmediately() = runBlockingTest {
-        val forwarder = SimResourceForwarder(SimCpu(1000.0))
+        val forwarder = SimResourceForwarder()
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(SimCpu(2000.0), clock, scheduler)
+        val source = SimResourceSource(2000.0, clock, scheduler)
 
         launch {
             source.consume(forwarder)
             source.close()
         }
 
-        forwarder.consume(object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+        forwarder.consume(object : SimResourceConsumer {
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand {
                 return SimResourceCommand.Exit
             }
         })
+
         forwarder.close()
         scheduler.close()
     }
 
     @Test
     fun testExit() = runBlockingTest {
-        val forwarder = SimResourceForwarder(SimCpu(1000.0))
+        val forwarder = SimResourceForwarder()
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val source = SimResourceSource(SimCpu(2000.0), clock, scheduler)
+        val source = SimResourceSource(2000.0, clock, scheduler)
 
         launch {
             source.consume(forwarder)
             source.close()
         }
 
-        forwarder.consume(object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1.0, 1.0)
-            }
+        forwarder.consume(object : SimResourceConsumer {
+            var isFirst = true
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+                return if (isFirst) {
+                    isFirst = false
+                    SimResourceCommand.Consume(10.0, 1.0)
+                } else {
+                    SimResourceCommand.Exit
+                }
             }
         })
 
         forwarder.close()
     }
+
+    @Test
+    fun testState() = runBlockingTest {
+        val forwarder = SimResourceForwarder()
+        val consumer = object : SimResourceConsumer {
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand = SimResourceCommand.Exit
+        }
+
+        assertEquals(SimResourceState.Pending, forwarder.state)
+
+        forwarder.startConsumer(consumer)
+        assertEquals(SimResourceState.Active, forwarder.state)
+
+        assertThrows<IllegalStateException> { forwarder.startConsumer(consumer) }
+
+        forwarder.cancel()
+        assertEquals(SimResourceState.Pending, forwarder.state)
+
+        forwarder.close()
+        assertEquals(SimResourceState.Stopped, forwarder.state)
+    }
+
+    @Test
+    fun testCancelPendingDelegate() = runBlockingTest {
+        val forwarder = SimResourceForwarder()
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Exit
+
+        forwarder.startConsumer(consumer)
+        forwarder.cancel()
+
+        verify(exactly = 0) { consumer.onFinish(any(), null) }
+    }
+
+    @Test
+    fun testCancelStartedDelegate() = runBlockingTest {
+        val forwarder = SimResourceForwarder()
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, clock, scheduler)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10)
+
+        source.startConsumer(forwarder)
+        yield()
+        forwarder.startConsumer(consumer)
+        yield()
+        forwarder.cancel()
+
+        verify(exactly = 1) { consumer.onStart(any()) }
+        verify(exactly = 1) { consumer.onFinish(any(), null) }
+    }
+
+    @Test
+    fun testCancelPropagation() = runBlockingTest {
+        val forwarder = SimResourceForwarder()
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, clock, scheduler)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10)
+
+        source.startConsumer(forwarder)
+        yield()
+        forwarder.startConsumer(consumer)
+        yield()
+        source.cancel()
+
+        verify(exactly = 1) { consumer.onStart(any()) }
+        verify(exactly = 1) { consumer.onFinish(any(), null) }
+    }
+
+    @Test
+    fun testAdjustCapacity() = runBlockingTest {
+        val forwarder = SimResourceForwarder()
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(1.0, clock, scheduler)
+
+        val consumer = spyk(SimWorkConsumer(2.0, 1.0))
+        source.startConsumer(forwarder)
+
+        coroutineScope {
+            launch { forwarder.consume(consumer) }
+            delay(1000)
+            source.capacity = 0.5
+        }
+
+        assertEquals(3000, currentTime)
+        verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
+    }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
index 4f7825fc..58e19421 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
@@ -22,11 +22,16 @@
 
 package org.opendc.simulator.resources
 
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.spyk
+import io.mockk.verify
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
 
@@ -35,26 +40,17 @@ import org.opendc.utils.TimerScheduler
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 class SimResourceSourceTest {
-    data class SimCpu(val speed: Double) : SimResource {
-        override val capacity: Double
-            get() = speed
-    }
-
     @Test
     fun testSpeed() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1000 * ctx.resource.speed, ctx.resource.speed)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1000 * capacity, capacity))
+            .andThen(SimResourceCommand.Exit)
 
         try {
             val res = mutableListOf<Double>()
@@ -63,7 +59,7 @@ class SimResourceSourceTest {
             provider.consume(consumer)
 
             job.cancel()
-            assertEquals(listOf(0.0, provider.resource.speed, 0.0), res) { "Speed is reported correctly" }
+            assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
         } finally {
             scheduler.close()
             provider.close()
@@ -71,20 +67,38 @@ class SimResourceSourceTest {
     }
 
     @Test
-    fun testSpeedLimit() = runBlockingTest {
+    fun testAdjustCapacity() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val provider = SimResourceSource(1.0, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1000 * ctx.resource.speed, 2 * ctx.resource.speed)
-            }
+        val consumer = spyk(SimWorkConsumer(2.0, 1.0))
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
+        try {
+            coroutineScope {
+                launch { provider.consume(consumer) }
+                delay(1000)
+                provider.capacity = 0.5
             }
+            assertEquals(3000, currentTime)
+            verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
+    }
+
+    @Test
+    fun testSpeedLimit() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1000 * capacity, 2 * capacity))
+            .andThen(SimResourceCommand.Exit)
 
         try {
             val res = mutableListOf<Double>()
@@ -93,7 +107,7 @@ class SimResourceSourceTest {
             provider.consume(consumer)
 
             job.cancel()
-            assertEquals(listOf(0.0, provider.resource.speed, 0.0), res) { "Speed is reported correctly" }
+            assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
         } finally {
             scheduler.close()
             provider.close()
@@ -108,16 +122,16 @@ class SimResourceSourceTest {
     fun testIntermediateInterrupt() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+        val consumer = object : SimResourceConsumer {
+            override fun onStart(ctx: SimResourceContext) {
                 ctx.interrupt()
-                return SimResourceCommand.Exit
             }
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                throw IllegalStateException()
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+                return SimResourceCommand.Exit
             }
         }
 
@@ -133,18 +147,24 @@ class SimResourceSourceTest {
     fun testInterrupt() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
-        lateinit var resCtx: SimResourceContext<SimCpu>
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
+        lateinit var resCtx: SimResourceContext
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+        val consumer = object : SimResourceConsumer {
+            var isFirst = true
+            override fun onStart(ctx: SimResourceContext) {
                 resCtx = ctx
-                return SimResourceCommand.Consume(4.0, 1.0)
             }
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                assertEquals(0.0, remainingWork)
-                return SimResourceCommand.Exit
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+                assertEquals(0.0, ctx.remainingWork)
+                return if (isFirst) {
+                    isFirst = false
+                    SimResourceCommand.Consume(4.0, 1.0)
+                } else {
+                    SimResourceCommand.Exit
+                }
             }
         }
 
@@ -166,17 +186,12 @@ class SimResourceSourceTest {
     fun testFailure() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                throw IllegalStateException()
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                throw IllegalStateException()
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onStart(any()) }
+            .throws(IllegalStateException())
 
         try {
             assertThrows<IllegalStateException> {
@@ -192,17 +207,13 @@ class SimResourceSourceTest {
     fun testExceptionPropagationOnNext() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1.0, 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                throw IllegalStateException()
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1.0, 1.0))
+            .andThenThrows(IllegalStateException())
 
         try {
             assertThrows<IllegalStateException> {
@@ -218,17 +229,13 @@ class SimResourceSourceTest {
     fun testConcurrentConsumption() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1.0, 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                throw IllegalStateException()
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1.0, 1.0))
+            .andThenThrows(IllegalStateException())
 
         try {
             assertThrows<IllegalStateException> {
@@ -247,17 +254,13 @@ class SimResourceSourceTest {
     fun testClosedConsumption() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1.0, 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                throw IllegalStateException()
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1.0, 1.0))
+            .andThenThrows(IllegalStateException())
 
         try {
             assertThrows<IllegalStateException> {
@@ -274,17 +277,13 @@ class SimResourceSourceTest {
     fun testCloseDuringConsumption() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1.0, 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                throw IllegalStateException()
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(1.0, 1.0))
+            .andThenThrows(IllegalStateException())
 
         try {
             launch { provider.consume(consumer) }
@@ -302,17 +301,13 @@ class SimResourceSourceTest {
     fun testIdle() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
 
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Idle(ctx.clock.millis() + 500)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Idle(clock.millis() + 500))
+            .andThen(SimResourceCommand.Exit)
 
         try {
             provider.consume(consumer)
@@ -330,17 +325,13 @@ class SimResourceSourceTest {
             runBlockingTest {
                 val clock = DelayControllerClockAdapter(this)
                 val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-                val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
-
-                val consumer = object : SimResourceConsumer<SimCpu> {
-                    override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                        return SimResourceCommand.Idle()
-                    }
+                val capacity = 4200.0
+                val provider = SimResourceSource(capacity, clock, scheduler)
 
-                    override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                        return SimResourceCommand.Exit
-                    }
-                }
+                val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+                every { consumer.onNext(any()) }
+                    .returns(SimResourceCommand.Idle())
+                    .andThenThrows(IllegalStateException())
 
                 try {
                     provider.consume(consumer)
@@ -351,4 +342,26 @@ class SimResourceSourceTest {
             }
         }
     }
+
+    @Test
+    fun testIncorrectDeadline() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val capacity = 4200.0
+        val provider = SimResourceSource(capacity, clock, scheduler)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Idle(2))
+            .andThen(SimResourceCommand.Exit)
+
+        try {
+            delay(10)
+
+            assertThrows<IllegalArgumentException> { provider.consume(consumer) }
+        } finally {
+            scheduler.close()
+            provider.close()
+        }
+    }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
index ca6558bf..edd60502 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
@@ -22,6 +22,8 @@
 
 package org.opendc.simulator.resources
 
+import io.mockk.every
+import io.mockk.mockk
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
@@ -34,18 +36,12 @@ import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.resources.consumer.SimTraceConsumer
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
-import java.lang.IllegalStateException
 
 /**
  * Test suite for the [SimResourceSwitchExclusive] class.
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class SimResourceSwitchExclusiveTest {
-    class SimCpu(val speed: Double) : SimResource {
-        override val capacity: Double
-            get() = speed
-    }
-
     /**
      * Test a trace workload.
      */
@@ -67,12 +63,12 @@ internal class SimResourceSwitchExclusiveTest {
                 ),
             )
 
-        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
-        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+        val switch = SimResourceSwitchExclusive()
+        val source = SimResourceSource(3200.0, clock, scheduler)
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(SimCpu(3200.0))
+        val provider = switch.addOutput(3200.0)
         val job = launch { source.speed.toList(speed) }
 
         try {
@@ -98,22 +94,15 @@ internal class SimResourceSwitchExclusiveTest {
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
-        val workload = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(duration / 1000.0, 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
+        val workload = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { workload.onNext(any()) } returns SimResourceCommand.Consume(duration / 1000.0, 1.0) andThen SimResourceCommand.Exit
 
-        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
-        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+        val switch = SimResourceSwitchExclusive()
+        val source = SimResourceSource(3200.0, clock, scheduler)
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(SimCpu(3200.0))
+        val provider = switch.addOutput(3200.0)
 
         try {
             provider.consume(workload)
@@ -133,22 +122,29 @@ internal class SimResourceSwitchExclusiveTest {
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
-        val workload = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(duration / 1000.0, 1.0)
+        val workload = object : SimResourceConsumer {
+            var isFirst = true
+
+            override fun onStart(ctx: SimResourceContext) {
+                isFirst = true
             }
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+                return if (isFirst) {
+                    isFirst = false
+                    SimResourceCommand.Consume(duration / 1000.0, 1.0)
+                } else {
+                    SimResourceCommand.Exit
+                }
             }
         }
 
-        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
-        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+        val switch = SimResourceSwitchExclusive()
+        val source = SimResourceSource(3200.0, clock, scheduler)
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(SimCpu(3200.0))
+        val provider = switch.addOutput(3200.0)
 
         try {
             provider.consume(workload)
@@ -169,22 +165,15 @@ internal class SimResourceSwitchExclusiveTest {
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
-        val workload = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(duration.toDouble(), 1.0)
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
+        val workload = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { workload.onNext(any()) } returns SimResourceCommand.Consume(duration / 1000.0, 1.0) andThen SimResourceCommand.Exit
 
-        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
-        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+        val switch = SimResourceSwitchExclusive()
+        val source = SimResourceSource(3200.0, clock, scheduler)
 
         switch.addInput(source)
 
-        switch.addOutput(SimCpu(3200.0))
-        assertThrows<IllegalStateException> { switch.addOutput(SimCpu(3200.0)) }
+        switch.addOutput(3200.0)
+        assertThrows<IllegalStateException> { switch.addOutput(3200.0) }
     }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
index 698c1700..5f4fd187 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
@@ -22,6 +22,8 @@
 
 package org.opendc.simulator.resources
 
+import io.mockk.every
+import io.mockk.mockk
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.launch
@@ -38,31 +40,19 @@ import org.opendc.utils.TimerScheduler
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class SimResourceSwitchMaxMinTest {
-    class SimCpu(val speed: Double) : SimResource {
-        override val capacity: Double
-            get() = speed
-    }
-
     @Test
     fun testSmoke() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
-        val switch = SimResourceSwitchMaxMin<SimCpu>(clock, coroutineContext)
+        val switch = SimResourceSwitchMaxMin(clock)
 
-        val sources = List(2) { SimResourceSource(SimCpu(2000.0), clock, scheduler) }
+        val sources = List(2) { SimResourceSource(2000.0, clock, scheduler) }
         sources.forEach { switch.addInput(it) }
 
-        val provider = switch.addOutput(SimCpu(1000.0))
-
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Consume(1.0, 1.0)
-            }
+        val provider = switch.addOutput(1000.0)
 
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Consume(1.0, 1.0) andThen SimResourceCommand.Exit
 
         try {
             provider.consume(consumer)
@@ -81,13 +71,13 @@ internal class SimResourceSwitchMaxMinTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
 
-        val listener = object : SimResourceSwitchMaxMin.Listener<SimCpu> {
+        val listener = object : SimResourceSwitchMaxMin.Listener {
             var totalRequestedWork = 0L
             var totalGrantedWork = 0L
             var totalOvercommittedWork = 0L
 
             override fun onSliceFinish(
-                switch: SimResourceSwitchMaxMin<SimCpu>,
+                switch: SimResourceSwitchMaxMin,
                 requestedWork: Long,
                 grantedWork: Long,
                 overcommittedWork: Long,
@@ -112,11 +102,11 @@ internal class SimResourceSwitchMaxMinTest {
                 ),
             )
 
-        val switch = SimResourceSwitchMaxMin(clock, coroutineContext, listener)
-        val provider = switch.addOutput(SimCpu(3200.0))
+        val switch = SimResourceSwitchMaxMin(clock, listener)
+        val provider = switch.addOutput(3200.0)
 
         try {
-            switch.addInput(SimResourceSource(SimCpu(3200.0), clock, scheduler))
+            switch.addInput(SimResourceSource(3200.0, clock, scheduler))
             provider.consume(workload)
             yield()
         } finally {
@@ -140,13 +130,13 @@ internal class SimResourceSwitchMaxMinTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)
 
-        val listener = object : SimResourceSwitchMaxMin.Listener<SimCpu> {
+        val listener = object : SimResourceSwitchMaxMin.Listener {
             var totalRequestedWork = 0L
             var totalGrantedWork = 0L
             var totalOvercommittedWork = 0L
 
             override fun onSliceFinish(
-                switch: SimResourceSwitchMaxMin<SimCpu>,
+                switch: SimResourceSwitchMaxMin,
                 requestedWork: Long,
                 grantedWork: Long,
                 overcommittedWork: Long,
@@ -180,12 +170,12 @@ internal class SimResourceSwitchMaxMinTest {
                 )
             )
 
-        val switch = SimResourceSwitchMaxMin(clock, coroutineContext, listener)
-        val providerA = switch.addOutput(SimCpu(3200.0))
-        val providerB = switch.addOutput(SimCpu(3200.0))
+        val switch = SimResourceSwitchMaxMin(clock, listener)
+        val providerA = switch.addOutput(3200.0)
+        val providerB = switch.addOutput(3200.0)
 
         try {
-            switch.addInput(SimResourceSource(SimCpu(3200.0), clock, scheduler))
+            switch.addInput(SimResourceSource(3200.0, clock, scheduler))
 
             coroutineScope {
                 launch { providerA.consume(workloadA) }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt
new file mode 100644
index 00000000..4d6b19ee
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimWorkConsumerTest.kt
@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.simulator.resources
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runBlockingTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+import org.opendc.utils.TimerScheduler
+
+/**
+ * A test suite for the [SimWorkConsumer] class.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class SimWorkConsumerTest {
+    @Test
+    fun testSmoke() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(1.0, clock, scheduler)
+
+        val consumer = SimWorkConsumer(1.0, 1.0)
+
+        try {
+            provider.consume(consumer)
+            assertEquals(1000, currentTime)
+        } finally {
+            provider.close()
+        }
+    }
+
+    @Test
+    fun testUtilization() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(1.0, clock, scheduler)
+
+        val consumer = SimWorkConsumer(1.0, 0.5)
+
+        try {
+            provider.consume(consumer)
+            assertEquals(2000, currentTime)
+        } finally {
+            provider.close()
+        }
+    }
+}