simulator: Remove concept of resource lifecycle

This change removes the AutoCloseable interface from the
SimResourceProvider and removes the concept of a resource lifecycle.
Instead, resource providers are now either active (running a resource
consumer) or in-active (being idle), which simplifies implementation.

17 files changed, 183 insertions, 283 deletions

diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
index d24ed1f3..c560cd28 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
@@ -129,6 +129,10 @@ public abstract class SimAbstractHypervisor(
         override fun close() {
             super.close()
 
+            for (cpu in cpus) {
+                cpu.close()
+            }
+
             _vms.remove(this)
         }
     }
@@ -137,9 +141,9 @@ public abstract class SimAbstractHypervisor(
      * A [SimProcessingUnit] of a virtual machine.
      */
     private class VCpu(
-        private val source: SimResourceProvider,
+        private val source: SimResourceCloseableProvider,
         override val model: ProcessingUnit
-    ) : SimProcessingUnit, SimResourceProvider by source {
+    ) : SimProcessingUnit, SimResourceCloseableProvider by source {
         override var capacity: Double
             get() = source.capacity
             set(_) {
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
index 93d306cf..3a70680c 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
@@ -115,11 +115,6 @@ public abstract class SimAbstractMachine(
 
         isTerminated = true
         cancel()
-        interpreter.batch {
-            for (cpu in cpus) {
-                cpu.close()
-            }
-        }
     }
 
     /* SimResourceSystem */
diff --git a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt b/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt
index 11034a57..3ce85d02 100644
--- a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt
+++ b/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt
@@ -83,7 +83,7 @@ public class SimPdu(
     /**
      * A PDU outlet.
      */
-    public class Outlet(private val provider: SimResourceProvider) : SimPowerOutlet(), AutoCloseable {
+    public class Outlet(private val provider: SimResourceCloseableProvider) : SimPowerOutlet(), AutoCloseable {
         override fun onConnect(inlet: SimPowerInlet) {
             provider.startConsumer(inlet.createConsumer())
         }
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
index 5fe7d7bb..84217278 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
@@ -56,8 +56,6 @@ public abstract class SimAbstractResourceAggregator(
 
     /* SimResourceAggregator */
     override fun addInput(input: SimResourceProvider) {
-        check(state != SimResourceState.Stopped) { "Aggregator has been stopped" }
-
         val consumer = Consumer()
         _inputs.add(input)
         _inputConsumers.add(consumer)
@@ -70,8 +68,8 @@ public abstract class SimAbstractResourceAggregator(
     private val _inputConsumers = mutableListOf<Consumer>()
 
     /* SimResourceProvider */
-    override val state: SimResourceState
-        get() = _output.state
+    override val isActive: Boolean
+        get() = _output.isActive
 
     override val capacity: Double
         get() = _output.capacity
@@ -97,10 +95,6 @@ public abstract class SimAbstractResourceAggregator(
         _output.interrupt()
     }
 
-    override fun close() {
-        _output.close()
-    }
-
     private val _output = object : SimAbstractResourceProvider(interpreter, parent, initialCapacity = 0.0) {
         override fun createLogic(): SimResourceProviderLogic {
             return object : SimResourceProviderLogic {
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt
index de26f99e..c1b1450e 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt
@@ -33,6 +33,12 @@ public abstract class SimAbstractResourceProvider(
     initialCapacity: Double
 ) : SimResourceProvider {
     /**
+     * A flag to indicate that the resource provider is active.
+     */
+    public override val isActive: Boolean
+        get() = ctx != null
+
+    /**
      * The capacity of the resource.
      */
     public override var capacity: Double = initialCapacity
@@ -67,12 +73,6 @@ public abstract class SimAbstractResourceProvider(
         private set
 
     /**
-     * The state of the resource provider.
-     */
-    final override var state: SimResourceState = SimResourceState.Pending
-        private set
-
-    /**
      * Construct the [SimResourceProviderLogic] instance for a new consumer.
      */
     protected abstract fun createLogic(): SimResourceProviderLogic
@@ -96,21 +96,15 @@ public abstract class SimAbstractResourceProvider(
     }
 
     final override fun startConsumer(consumer: SimResourceConsumer) {
-        check(state == SimResourceState.Pending) { "Resource is in invalid state" }
+        check(ctx == null) { "Resource is in invalid state" }
         val ctx = interpreter.newContext(consumer, createLogic(), parent)
 
         ctx.capacity = capacity
         this.ctx = ctx
-        this.state = SimResourceState.Active
 
         start(ctx)
     }
 
-    override fun close() {
-        cancel()
-        state = SimResourceState.Stopped
-    }
-
     final override fun interrupt() {
         ctx?.interrupt()
     }
@@ -121,10 +115,6 @@ public abstract class SimAbstractResourceProvider(
             this.ctx = null
             ctx.close()
         }
-
-        if (state != SimResourceState.Stopped) {
-            state = SimResourceState.Pending
-        }
     }
 
     override fun toString(): String = "SimAbstractResourceProvider[capacity=$capacity]"
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceCloseableProvider.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceCloseableProvider.kt
new file mode 100644
index 00000000..bce8274b
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceCloseableProvider.kt
@@ -0,0 +1,37 @@
+/*
+ * 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
+
+/**
+ * A [SimResourceProvider] that has a controllable and limited lifetime.
+ *
+ * This interface is used to signal that the resource provider may be closed and not reused after that point.
+ */
+public interface SimResourceCloseableProvider : SimResourceProvider, AutoCloseable {
+    /**
+     * End the lifetime of the resource provider.
+     *
+     * This operation cancels the existing resource consumer and prevents the resource provider from being reused.
+     */
+    public override fun close()
+}
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributor.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributor.kt
index e0333ff9..6bfbfc99 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributor.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributor.kt
@@ -29,10 +29,10 @@ public interface SimResourceDistributor : SimResourceConsumer {
     /**
      * The output resource providers to which resource consumers can be attached.
      */
-    public val outputs: Set<SimResourceProvider>
+    public val outputs: Set<SimResourceCloseableProvider>
 
     /**
      * Create a new output for the distributor.
      */
-    public fun newOutput(): SimResourceProvider
+    public fun newOutput(): SimResourceCloseableProvider
 }
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
index be9e89fb..f7c5c5d7 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
@@ -32,7 +32,7 @@ public class SimResourceDistributorMaxMin(
     private val interpreter: SimResourceInterpreter,
     private val parent: SimResourceSystem? = null
 ) : SimResourceDistributor {
-    override val outputs: Set<SimResourceProvider>
+    override val outputs: Set<SimResourceCloseableProvider>
         get() = _outputs
     private val _outputs = mutableSetOf<Output>()
 
@@ -57,7 +57,7 @@ public class SimResourceDistributorMaxMin(
     private var totalAllocatedSpeed = 0.0
 
     /* SimResourceDistributor */
-    override fun newOutput(): SimResourceProvider {
+    override fun newOutput(): SimResourceCloseableProvider {
         val provider = Output(ctx?.capacity ?: 0.0)
         _outputs.add(provider)
         return provider
@@ -178,7 +178,16 @@ public class SimResourceDistributorMaxMin(
     /**
      * An internal [SimResourceProvider] implementation for switch outputs.
      */
-    private inner class Output(capacity: Double) : SimAbstractResourceProvider(interpreter, parent, capacity), SimResourceProviderLogic, Comparable<Output> {
+    private inner class Output(capacity: Double) :
+        SimAbstractResourceProvider(interpreter, parent, capacity),
+        SimResourceCloseableProvider,
+        SimResourceProviderLogic,
+        Comparable<Output> {
+        /**
+         * A flag to indicate that the output is closed.
+         */
+        private var isClosed: Boolean = false
+
         /**
          * The current command that is processed by the resource.
          */
@@ -209,6 +218,8 @@ public class SimResourceDistributorMaxMin(
         override fun createLogic(): SimResourceProviderLogic = this
 
         override fun start(ctx: SimResourceControllableContext) {
+            check(!isClosed) { "Cannot re-use closed output" }
+
             activeOutputs += this
 
             interpreter.batch {
@@ -219,13 +230,9 @@ public class SimResourceDistributorMaxMin(
         }
 
         override fun close() {
-            val state = state
-
-            super.close()
-
-            if (state != SimResourceState.Stopped) {
-                _outputs.remove(this)
-            }
+            isClosed = true
+            cancel()
+            _outputs.remove(this)
         }
 
         /* SimResourceProviderLogic */
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt
index f709ca17..b68b7261 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt
@@ -29,11 +29,11 @@ import kotlin.coroutines.resumeWithException
 /**
  * A [SimResourceProvider] provides a resource that can be consumed by a [SimResourceConsumer].
  */
-public interface SimResourceProvider : AutoCloseable {
+public interface SimResourceProvider {
     /**
-     * The state of the resource.
+     * A flag to indicate that the resource provider is currently being consumed by a [SimResourceConsumer].
      */
-    public val state: SimResourceState
+    public val isActive: Boolean
 
     /**
      * The resource capacity available at this instant.
@@ -71,13 +71,6 @@ public interface SimResourceProvider : AutoCloseable {
      * Cancel the current resource consumer. If there is no consumer active, this operation will be a no-op.
      */
     public fun cancel()
-
-    /**
-     * End the lifetime of the resource.
-     *
-     * This operation terminates the existing resource consumer.
-     */
-    public override fun close()
 }
 
 /**
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitch.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitch.kt
index e224285e..f6e7b22f 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitch.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitch.kt
@@ -29,7 +29,7 @@ public interface SimResourceSwitch : AutoCloseable {
     /**
      * The output resource providers to which resource consumers can be attached.
      */
-    public val outputs: Set<SimResourceProvider>
+    public val outputs: Set<SimResourceCloseableProvider>
 
     /**
      * The input resources that will be switched between the output providers.
@@ -44,7 +44,7 @@ public interface SimResourceSwitch : AutoCloseable {
     /**
      * Create a new output on the switch.
      */
-    public fun newOutput(): SimResourceProvider
+    public fun newOutput(): SimResourceCloseableProvider
 
     /**
      * Add the specified [input] to the switch.
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt
index 2950af80..4ff741ed 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt
@@ -35,7 +35,7 @@ public class SimResourceSwitchExclusive : SimResourceSwitch {
     private var isClosed: Boolean = false
 
     private val _outputs = mutableSetOf<Provider>()
-    override val outputs: Set<SimResourceProvider>
+    override val outputs: Set<SimResourceCloseableProvider>
         get() = _outputs
 
     private val availableResources = ArrayDeque<SimResourceTransformer>()
@@ -61,7 +61,7 @@ public class SimResourceSwitchExclusive : SimResourceSwitch {
         override fun toString(): String = "SimResourceCounters[demand=$demand,actual=$actual,overcommit=$overcommit]"
     }
 
-    override fun newOutput(): SimResourceProvider {
+    override fun newOutput(): SimResourceCloseableProvider {
         check(!isClosed) { "Switch has been closed" }
         check(availableResources.isNotEmpty()) { "No capacity to serve request" }
         val forwarder = availableResources.poll()
@@ -101,7 +101,7 @@ public class SimResourceSwitchExclusive : SimResourceSwitch {
         _inputs.forEach(SimResourceProvider::cancel)
     }
 
-    private inner class Provider(private val forwarder: SimResourceTransformer) : SimResourceProvider by forwarder {
+    private inner class Provider(private val forwarder: SimResourceTransformer) : SimResourceCloseableProvider, SimResourceProvider by forwarder {
         override fun close() {
             // We explicitly do not close the forwarder here in order to re-use it across output resources.
             _outputs -= this
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
index 684a1b52..50d58798 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
@@ -33,7 +33,7 @@ public class SimResourceSwitchMaxMin(
     /**
      * The output resource providers to which resource consumers can be attached.
      */
-    override val outputs: Set<SimResourceProvider>
+    override val outputs: Set<SimResourceCloseableProvider>
         get() = distributor.outputs
 
     /**
@@ -70,7 +70,7 @@ public class SimResourceSwitchMaxMin(
     /**
      * Add an output to the switch.
      */
-    override fun newOutput(): SimResourceProvider {
+    override fun newOutput(): SimResourceCloseableProvider {
         check(!isClosed) { "Switch has been closed" }
 
         return distributor.newOutput()
@@ -88,7 +88,7 @@ public class SimResourceSwitchMaxMin(
     override fun close() {
         if (!isClosed) {
             isClosed = true
-            aggregator.close()
+            aggregator.cancel()
         }
     }
 }
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt
index fd3d1230..cec27e1c 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt
@@ -33,7 +33,7 @@ import org.opendc.simulator.resources.impl.SimResourceCountersImpl
 public class SimResourceTransformer(
     private val isCoupled: Boolean = false,
     private val transform: (SimResourceContext, SimResourceCommand) -> SimResourceCommand
-) : SimResourceFlow {
+) : SimResourceFlow, AutoCloseable {
     /**
      * The [SimResourceContext] in which the forwarder runs.
      */
@@ -49,11 +49,8 @@ public class SimResourceTransformer(
      */
     private var hasDelegateStarted: Boolean = false
 
-    /**
-     * The state of the forwarder.
-     */
-    override var state: SimResourceState = SimResourceState.Pending
-        private set
+    override val isActive: Boolean
+        get() = delegate != null
 
     override val capacity: Double
         get() = ctx?.capacity ?: 0.0
@@ -69,9 +66,8 @@ public class SimResourceTransformer(
     private val _counters = SimResourceCountersImpl()
 
     override fun startConsumer(consumer: SimResourceConsumer) {
-        check(state == SimResourceState.Pending) { "Resource is in invalid state" }
+        check(delegate == null) { "Resource transformer already active" }
 
-        state = SimResourceState.Active
         delegate = consumer
 
         // Interrupt the provider to replace the consumer
@@ -86,19 +82,18 @@ public class SimResourceTransformer(
         val delegate = delegate
         val ctx = ctx
 
-        state = SimResourceState.Pending
-
-        if (delegate != null && ctx != null) {
+        if (delegate != null) {
             this.delegate = null
-            delegate.onEvent(ctx, SimResourceEvent.Exit)
+
+            if (ctx != null) {
+                delegate.onEvent(ctx, SimResourceEvent.Exit)
+            }
         }
     }
 
     override fun close() {
         val ctx = ctx
 
-        state = SimResourceState.Stopped
-
         if (ctx != null) {
             this.ctx = null
             ctx.interrupt()
@@ -114,9 +109,7 @@ public class SimResourceTransformer(
 
         updateCounters(ctx)
 
-        return if (state == SimResourceState.Stopped) {
-            SimResourceCommand.Exit
-        } else if (delegate != null) {
+        return if (delegate != null) {
             val command = transform(ctx, delegate.onNext(ctx))
 
             _work = if (command is SimResourceCommand.Consume) command.work else 0.0
@@ -128,7 +121,7 @@ public class SimResourceTransformer(
 
                 delegate.onEvent(ctx, SimResourceEvent.Exit)
 
-                if (isCoupled || state == SimResourceState.Stopped)
+                if (isCoupled)
                     SimResourceCommand.Exit
                 else
                     onNext(ctx)
@@ -184,10 +177,6 @@ public class SimResourceTransformer(
     private fun reset() {
         delegate = null
         hasDelegateStarted = false
-
-        if (state != SimResourceState.Stopped) {
-            state = SimResourceState.Pending
-        }
     }
 
     /**
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 51024e80..2f01a8c4 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
@@ -26,7 +26,7 @@ import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
 import kotlinx.coroutines.*
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
 import org.junit.jupiter.api.assertThrows
@@ -58,17 +58,13 @@ internal class SimResourceAggregatorMaxMinTest {
         val adapter = SimSpeedConsumerAdapter(forwarder, usage::add)
         source.startConsumer(adapter)
 
-        try {
-            aggregator.consume(consumer)
-            yield()
+        aggregator.consume(consumer)
+        yield()
 
-            assertAll(
-                { assertEquals(1000, clock.millis()) },
-                { assertEquals(listOf(0.0, 0.5, 0.0), usage) }
-            )
-        } finally {
-            aggregator.close()
-        }
+        assertAll(
+            { assertEquals(1000, clock.millis()) },
+            { assertEquals(listOf(0.0, 0.5, 0.0), usage) }
+        )
     }
 
     @Test
@@ -86,16 +82,12 @@ internal class SimResourceAggregatorMaxMinTest {
         val usage = mutableListOf<Double>()
         val adapter = SimSpeedConsumerAdapter(consumer, usage::add)
 
-        try {
-            aggregator.consume(adapter)
-            yield()
-            assertAll(
-                { assertEquals(1000, clock.millis()) },
-                { assertEquals(listOf(0.0, 2.0, 0.0), usage) }
-            )
-        } finally {
-            aggregator.close()
-        }
+        aggregator.consume(adapter)
+        yield()
+        assertAll(
+            { assertEquals(1000, clock.millis()) },
+            { assertEquals(listOf(0.0, 2.0, 0.0), usage) }
+        )
     }
 
     @Test
@@ -114,15 +106,11 @@ internal class SimResourceAggregatorMaxMinTest {
             .returns(SimResourceCommand.Consume(4.0, 4.0, 1000))
             .andThen(SimResourceCommand.Exit)
 
-        try {
-            aggregator.consume(consumer)
-            yield()
-            assertEquals(1000, clock.millis())
+        aggregator.consume(consumer)
+        yield()
+        assertEquals(1000, clock.millis())
 
-            verify(exactly = 2) { consumer.onNext(any()) }
-        } finally {
-            aggregator.close()
-        }
+        verify(exactly = 2) { consumer.onNext(any()) }
     }
 
     @Test
@@ -141,13 +129,9 @@ internal class SimResourceAggregatorMaxMinTest {
             .returns(SimResourceCommand.Consume(1.0, 1.0))
             .andThenThrows(IllegalStateException("Test Exception"))
 
-        try {
-            assertThrows<IllegalStateException> { aggregator.consume(consumer) }
-            yield()
-            assertEquals(SimResourceState.Pending, sources[0].state)
-        } finally {
-            aggregator.close()
-        }
+        assertThrows<IllegalStateException> { aggregator.consume(consumer) }
+        yield()
+        assertFalse(sources[0].isActive)
     }
 
     @Test
@@ -162,17 +146,13 @@ internal class SimResourceAggregatorMaxMinTest {
         sources.forEach(aggregator::addInput)
 
         val consumer = SimWorkConsumer(4.0, 1.0)
-        try {
-            coroutineScope {
-                launch { aggregator.consume(consumer) }
-                delay(1000)
-                sources[0].capacity = 0.5
-            }
-            yield()
-            assertEquals(2334, clock.millis())
-        } finally {
-            aggregator.close()
+        coroutineScope {
+            launch { aggregator.consume(consumer) }
+            delay(1000)
+            sources[0].capacity = 0.5
         }
+        yield()
+        assertEquals(2334, clock.millis())
     }
 
     @Test
@@ -187,17 +167,13 @@ internal class SimResourceAggregatorMaxMinTest {
         sources.forEach(aggregator::addInput)
 
         val consumer = SimWorkConsumer(1.0, 0.5)
-        try {
-            coroutineScope {
-                launch { aggregator.consume(consumer) }
-                delay(500)
-                sources[0].capacity = 0.5
-            }
-            yield()
-            assertEquals(1000, clock.millis())
-        } finally {
-            aggregator.close()
+        coroutineScope {
+            launch { aggregator.consume(consumer) }
+            delay(500)
+            sources[0].capacity = 0.5
         }
+        yield()
+        assertEquals(1000, clock.millis())
     }
 
     @Test
@@ -216,13 +192,9 @@ internal class SimResourceAggregatorMaxMinTest {
             .returns(SimResourceCommand.Consume(4.0, 4.0, 1000))
             .andThen(SimResourceCommand.Exit)
 
-        try {
-            aggregator.consume(consumer)
-            yield()
-            assertEquals(1000, clock.millis())
-            assertEquals(2.0, aggregator.counters.actual) { "Actual work mismatch" }
-        } finally {
-            aggregator.close()
-        }
+        aggregator.consume(consumer)
+        yield()
+        assertEquals(1000, clock.millis())
+        assertEquals(2.0, aggregator.counters.actual) { "Actual work mismatch" }
     }
 }
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 08d88093..4895544d 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
@@ -50,16 +50,12 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Consume(1000 * capacity, capacity))
             .andThen(SimResourceCommand.Exit)
 
-        try {
-            val res = mutableListOf<Double>()
-            val adapter = SimSpeedConsumerAdapter(consumer, res::add)
+        val res = mutableListOf<Double>()
+        val adapter = SimSpeedConsumerAdapter(consumer, res::add)
 
-            provider.consume(adapter)
+        provider.consume(adapter)
 
-            assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
-        } finally {
-            provider.close()
-        }
+        assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
     }
 
     @Test
@@ -69,17 +65,13 @@ class SimResourceSourceTest {
 
         val consumer = spyk(SimWorkConsumer(2.0, 1.0))
 
-        try {
-            coroutineScope {
-                launch { provider.consume(consumer) }
-                delay(1000)
-                provider.capacity = 0.5
-            }
-            assertEquals(3000, clock.millis())
-            verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Capacity) }
-        } finally {
-            provider.close()
+        coroutineScope {
+            launch { provider.consume(consumer) }
+            delay(1000)
+            provider.capacity = 0.5
         }
+        assertEquals(3000, clock.millis())
+        verify(exactly = 1) { consumer.onEvent(any(), SimResourceEvent.Capacity) }
     }
 
     @Test
@@ -93,16 +85,12 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Consume(1000 * capacity, 2 * capacity))
             .andThen(SimResourceCommand.Exit)
 
-        try {
-            val res = mutableListOf<Double>()
-            val adapter = SimSpeedConsumerAdapter(consumer, res::add)
+        val res = mutableListOf<Double>()
+        val adapter = SimSpeedConsumerAdapter(consumer, res::add)
 
-            provider.consume(adapter)
+        provider.consume(adapter)
 
-            assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
-        } finally {
-            provider.close()
-        }
+        assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" }
     }
 
     /**
@@ -125,11 +113,7 @@ class SimResourceSourceTest {
             }
         }
 
-        try {
-            provider.consume(consumer)
-        } finally {
-            provider.close()
-        }
+        provider.consume(consumer)
     }
 
     @Test
@@ -160,17 +144,13 @@ class SimResourceSourceTest {
             }
         }
 
-        try {
-            launch {
-                yield()
-                resCtx.interrupt()
-            }
-            provider.consume(consumer)
-
-            assertEquals(0, clock.millis())
-        } finally {
-            provider.close()
+        launch {
+            yield()
+            resCtx.interrupt()
         }
+        provider.consume(consumer)
+
+        assertEquals(0, clock.millis())
     }
 
     @Test
@@ -183,12 +163,8 @@ class SimResourceSourceTest {
         every { consumer.onEvent(any(), eq(SimResourceEvent.Start)) }
             .throws(IllegalStateException())
 
-        try {
-            assertThrows<IllegalStateException> {
-                provider.consume(consumer)
-            }
-        } finally {
-            provider.close()
+        assertThrows<IllegalStateException> {
+            provider.consume(consumer)
         }
     }
 
@@ -203,12 +179,8 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Consume(1.0, 1.0))
             .andThenThrows(IllegalStateException())
 
-        try {
-            assertThrows<IllegalStateException> {
-                provider.consume(consumer)
-            }
-        } finally {
-            provider.close()
+        assertThrows<IllegalStateException> {
+            provider.consume(consumer)
         }
     }
 
@@ -223,41 +195,16 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Consume(1.0, 1.0))
             .andThenThrows(IllegalStateException())
 
-        try {
-            assertThrows<IllegalStateException> {
-                coroutineScope {
-                    launch { provider.consume(consumer) }
-                    provider.consume(consumer)
-                }
-            }
-        } finally {
-            provider.close()
-        }
-    }
-
-    @Test
-    fun testClosedConsumption() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-        val capacity = 4200.0
-        val provider = SimResourceSource(capacity, scheduler)
-
-        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
-        every { consumer.onNext(any()) }
-            .returns(SimResourceCommand.Consume(1.0, 1.0))
-            .andThenThrows(IllegalStateException())
-
-        try {
-            assertThrows<IllegalStateException> {
-                provider.close()
+        assertThrows<IllegalStateException> {
+            coroutineScope {
+                launch { provider.consume(consumer) }
                 provider.consume(consumer)
             }
-        } finally {
-            provider.close()
         }
     }
 
     @Test
-    fun testCloseDuringConsumption() = runBlockingSimulation {
+    fun testCancelDuringConsumption() = runBlockingSimulation {
         val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = SimResourceSource(capacity, scheduler)
@@ -267,15 +214,11 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Consume(1.0, 1.0))
             .andThenThrows(IllegalStateException())
 
-        try {
-            launch { provider.consume(consumer) }
-            delay(500)
-            provider.close()
+        launch { provider.consume(consumer) }
+        delay(500)
+        provider.cancel()
 
-            assertEquals(500, clock.millis())
-        } finally {
-            provider.close()
-        }
+        assertEquals(500, clock.millis())
     }
 
     @Test
@@ -289,13 +232,9 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Idle(clock.millis() + 500))
             .andThen(SimResourceCommand.Exit)
 
-        try {
-            provider.consume(consumer)
+        provider.consume(consumer)
 
-            assertEquals(500, clock.millis())
-        } finally {
-            provider.close()
-        }
+        assertEquals(500, clock.millis())
     }
 
     @Test
@@ -311,11 +250,7 @@ class SimResourceSourceTest {
                     .returns(SimResourceCommand.Idle())
                     .andThenThrows(IllegalStateException())
 
-                try {
-                    provider.consume(consumer)
-                } finally {
-                    provider.close()
-                }
+                provider.consume(consumer)
             }
         }
     }
@@ -331,12 +266,8 @@ class SimResourceSourceTest {
             .returns(SimResourceCommand.Idle(2))
             .andThen(SimResourceCommand.Exit)
 
-        try {
-            delay(10)
+        delay(10)
 
-            assertThrows<IllegalArgumentException> { provider.consume(consumer) }
-        } finally {
-            provider.close()
-        }
+        assertThrows<IllegalArgumentException> { provider.consume(consumer) }
     }
 }
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 810052b8..cf69b7b5 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
@@ -40,15 +40,12 @@ import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class SimResourceTransformerTest {
     @Test
-    fun testExitImmediately() = runBlockingSimulation {
+    fun testCancelImmediately() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
         val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
-        launch {
-            source.consume(forwarder)
-            source.close()
-        }
+        launch { source.consume(forwarder) }
 
         forwarder.consume(object : SimResourceConsumer {
             override fun onNext(ctx: SimResourceContext): SimResourceCommand {
@@ -57,18 +54,16 @@ internal class SimResourceTransformerTest {
         })
 
         forwarder.close()
+        source.cancel()
     }
 
     @Test
-    fun testExit() = runBlockingSimulation {
+    fun testCancel() = runBlockingSimulation {
         val forwarder = SimResourceForwarder()
         val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
         val source = SimResourceSource(2000.0, scheduler)
 
-        launch {
-            source.consume(forwarder)
-            source.close()
-        }
+        launch { source.consume(forwarder) }
 
         forwarder.consume(object : SimResourceConsumer {
             var isFirst = true
@@ -84,6 +79,7 @@ internal class SimResourceTransformerTest {
         })
 
         forwarder.close()
+        source.cancel()
     }
 
     @Test
@@ -93,18 +89,18 @@ internal class SimResourceTransformerTest {
             override fun onNext(ctx: SimResourceContext): SimResourceCommand = SimResourceCommand.Exit
         }
 
-        assertEquals(SimResourceState.Pending, forwarder.state)
+        assertFalse(forwarder.isActive)
 
         forwarder.startConsumer(consumer)
-        assertEquals(SimResourceState.Active, forwarder.state)
+        assertTrue(forwarder.isActive)
 
         assertThrows<IllegalStateException> { forwarder.startConsumer(consumer) }
 
         forwarder.cancel()
-        assertEquals(SimResourceState.Pending, forwarder.state)
+        assertFalse(forwarder.isActive)
 
         forwarder.close()
-        assertEquals(SimResourceState.Stopped, forwarder.state)
+        assertFalse(forwarder.isActive)
     }
 
     @Test
@@ -171,7 +167,7 @@ internal class SimResourceTransformerTest {
         forwarder.consume(consumer)
         yield()
 
-        assertEquals(SimResourceState.Pending, source.state)
+        assertFalse(forwarder.isActive)
     }
 
     @Test
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 db4fe856..42648cf1 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
@@ -41,12 +41,8 @@ internal class SimWorkConsumerTest {
 
         val consumer = SimWorkConsumer(1.0, 1.0)
 
-        try {
-            provider.consume(consumer)
-            assertEquals(1000, clock.millis())
-        } finally {
-            provider.close()
-        }
+        provider.consume(consumer)
+        assertEquals(1000, clock.millis())
     }
 
     @Test
@@ -56,11 +52,7 @@ internal class SimWorkConsumerTest {
 
         val consumer = SimWorkConsumer(1.0, 0.5)
 
-        try {
-            provider.consume(consumer)
-            assertEquals(2000, clock.millis())
-        } finally {
-            provider.close()
-        }
+        provider.consume(consumer)
+        assertEquals(2000, clock.millis())
     }
 }