merge: Extract scheduler from simulation coroutine dispatcher (#106)

This pull request extracts the scheduler from the `SimulationCoroutineDispatcher` into 
a separate `SimulationScheduler` class which allows users to re-use the scheduler 
between different coroutine dispatchers.

We implement the `SimulationScheduler` in Java, removing the explicit dependency on 
Kotlin or `kotlinx-coroutines`. The scheduler uses a separate specialized priority queue 
implementation that eliminates allocation in the hot path of the simulator.

## Implementation Notes :hammer_and_pick:

* Add Java-based simulator core
* Use SimulationScheduler in coroutine dispatcher
* Rename runBlockingSimulation to runSimulation

## External Dependencies :four_leaf_clover:

* N/A 

## Breaking API Changes :warning:

* The Kotlin API for simulation has been moved to `org.opendc.simulator.kotlin`.
* `runBlockingSImulation` renamed to `runSimulation`

43 files changed, 1204 insertions, 392 deletions

diff --git a/opendc-common/src/test/kotlin/org/opendc/common/util/PacerTest.kt b/opendc-common/src/test/kotlin/org/opendc/common/util/PacerTest.kt
index 1cd435f6..de9fd472 100644
--- a/opendc-common/src/test/kotlin/org/opendc/common/util/PacerTest.kt
+++ b/opendc-common/src/test/kotlin/org/opendc/common/util/PacerTest.kt
@@ -26,7 +26,7 @@ import kotlinx.coroutines.delay
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.time.Clock
 import kotlin.coroutines.EmptyCoroutineContext
 
@@ -43,7 +43,7 @@ class PacerTest {
     fun testSingleEnqueue() {
         var count = 0
 
-        runBlockingSimulation {
+        runSimulation {
             val pacer = Pacer(coroutineContext, clock, quantum = 100) {
                 count++
             }
@@ -58,7 +58,7 @@ class PacerTest {
     fun testCascade() {
         var count = 0
 
-        runBlockingSimulation {
+        runSimulation {
             val pacer = Pacer(coroutineContext, clock, quantum = 100) {
                 count++
             }
@@ -76,7 +76,7 @@ class PacerTest {
     fun testCancel() {
         var count = 0
 
-        runBlockingSimulation {
+        runSimulation {
             val pacer = Pacer(coroutineContext, clock, quantum = 100) {
                 count++
             }
@@ -94,7 +94,7 @@ class PacerTest {
     fun testCancelWithoutPending() {
         var count = 0
 
-        runBlockingSimulation {
+        runSimulation {
             val pacer = Pacer(coroutineContext, clock, quantum = 100) {
                 count++
             }
@@ -112,7 +112,7 @@ class PacerTest {
     fun testSubsequent() {
         var count = 0
 
-        runBlockingSimulation {
+        runSimulation {
             val pacer = Pacer(coroutineContext, clock, quantum = 100) {
                 count++
             }
diff --git a/opendc-common/src/test/kotlin/org/opendc/common/util/TimerSchedulerTest.kt b/opendc-common/src/test/kotlin/org/opendc/common/util/TimerSchedulerTest.kt
index 01f61f92..183ab66a 100644
--- a/opendc-common/src/test/kotlin/org/opendc/common/util/TimerSchedulerTest.kt
+++ b/opendc-common/src/test/kotlin/org/opendc/common/util/TimerSchedulerTest.kt
@@ -26,7 +26,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.time.Clock
 import kotlin.coroutines.EmptyCoroutineContext
 
@@ -42,7 +42,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testBasicTimer() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             scheduler.startSingleTimer(0, 1000) {
@@ -56,7 +56,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testCancelNonExisting() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             scheduler.cancel(1)
@@ -65,7 +65,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testCancelExisting() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             scheduler.startSingleTimer(0, 1000) {
@@ -82,7 +82,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testCancelAll() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             scheduler.startSingleTimer(0, 1000) { fail() }
@@ -93,7 +93,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testOverride() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             scheduler.startSingleTimer(0, 1000) { fail() }
@@ -107,7 +107,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testOverrideBlock() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             scheduler.startSingleTimer(0, 1000) { fail() }
@@ -120,7 +120,7 @@ internal class TimerSchedulerTest {
 
     @Test
     fun testNegativeDelay() {
-        runBlockingSimulation {
+        runSimulation {
             val scheduler = TimerScheduler<Int>(coroutineContext, clock)
 
             assertThrows<IllegalArgumentException> {
diff --git a/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt b/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt
index cc7be4a8..4f4008bc 100644
--- a/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt
+++ b/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/ComputeServiceTest.kt
@@ -39,8 +39,8 @@ import org.opendc.compute.service.scheduler.filters.ComputeFilter
 import org.opendc.compute.service.scheduler.filters.RamFilter
 import org.opendc.compute.service.scheduler.filters.VCpuFilter
 import org.opendc.compute.service.scheduler.weights.RamWeigher
-import org.opendc.simulator.core.SimulationCoroutineScope
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.SimulationCoroutineScope
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -62,7 +62,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testClientClose() = scope.runBlockingSimulation {
+    fun testClientClose() = scope.runSimulation {
         val client = service.newClient()
 
         assertEquals(emptyList<Flavor>(), client.queryFlavors())
@@ -85,7 +85,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testClientCreate() = scope.runBlockingSimulation {
+    fun testClientCreate() = scope.runSimulation {
         val client = service.newClient()
 
         val flavor = client.newFlavor("test", 1, 1024)
@@ -111,7 +111,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testClientOnClose() = scope.runBlockingSimulation {
+    fun testClientOnClose() = scope.runSimulation {
         service.close()
         assertThrows<IllegalStateException> {
             service.newClient()
@@ -119,7 +119,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testAddHost() = scope.runBlockingSimulation {
+    fun testAddHost() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
 
         every { host.model } returns HostModel(4 * 2600.0, 4, 2048)
@@ -139,7 +139,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testAddHostDouble() = scope.runBlockingSimulation {
+    fun testAddHostDouble() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
 
         every { host.model } returns HostModel(4 * 2600.0, 4, 2048)
@@ -154,7 +154,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerStartWithoutEnoughCpus() = scope.runBlockingSimulation {
+    fun testServerStartWithoutEnoughCpus() = scope.runSimulation {
         val client = service.newClient()
         val flavor = client.newFlavor("test", 1, 0)
         val image = client.newImage("test")
@@ -167,7 +167,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerStartWithoutEnoughMemory() = scope.runBlockingSimulation {
+    fun testServerStartWithoutEnoughMemory() = scope.runSimulation {
         val client = service.newClient()
         val flavor = client.newFlavor("test", 0, 1024)
         val image = client.newImage("test")
@@ -180,7 +180,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerStartWithoutEnoughResources() = scope.runBlockingSimulation {
+    fun testServerStartWithoutEnoughResources() = scope.runSimulation {
         val client = service.newClient()
         val flavor = client.newFlavor("test", 1, 1024)
         val image = client.newImage("test")
@@ -193,7 +193,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerCancelRequest() = scope.runBlockingSimulation {
+    fun testServerCancelRequest() = scope.runSimulation {
         val client = service.newClient()
         val flavor = client.newFlavor("test", 1, 1024)
         val image = client.newImage("test")
@@ -207,7 +207,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerCannotFitOnHost() = scope.runBlockingSimulation {
+    fun testServerCannotFitOnHost() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
 
         every { host.model } returns HostModel(4 * 2600.0, 4, 2048)
@@ -230,7 +230,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testHostAvailableAfterSomeTime() = scope.runBlockingSimulation {
+    fun testHostAvailableAfterSomeTime() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
         val listeners = mutableListOf<HostListener>()
 
@@ -261,7 +261,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testHostUnavailableAfterSomeTime() = scope.runBlockingSimulation {
+    fun testHostUnavailableAfterSomeTime() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
         val listeners = mutableListOf<HostListener>()
 
@@ -292,7 +292,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerInvalidType() = scope.runBlockingSimulation {
+    fun testServerInvalidType() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
         val listeners = mutableListOf<HostListener>()
 
@@ -315,7 +315,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerDeploy() = scope.runBlockingSimulation {
+    fun testServerDeploy() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
         val listeners = mutableListOf<HostListener>()
 
@@ -358,7 +358,7 @@ internal class ComputeServiceTest {
     }
 
     @Test
-    fun testServerDeployFailure() = scope.runBlockingSimulation {
+    fun testServerDeployFailure() = scope.runSimulation {
         val host = mockk<Host>(relaxUnitFun = true)
         val listeners = mutableListOf<HostListener>()
 
diff --git a/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/InternalServerTest.kt b/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/InternalServerTest.kt
index 4e5a37ae..9e59949f 100644
--- a/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/InternalServerTest.kt
+++ b/opendc-compute/opendc-compute-service/src/test/kotlin/org/opendc/compute/service/InternalServerTest.kt
@@ -34,7 +34,7 @@ import org.opendc.compute.service.internal.ComputeServiceImpl
 import org.opendc.compute.service.internal.InternalFlavor
 import org.opendc.compute.service.internal.InternalImage
 import org.opendc.compute.service.internal.InternalServer
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -94,7 +94,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStartTerminatedServer() = runBlockingSimulation {
+    fun testStartTerminatedServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -110,7 +110,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStartDeletedServer() = runBlockingSimulation {
+    fun testStartDeletedServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -123,7 +123,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStartProvisioningServer() = runBlockingSimulation {
+    fun testStartProvisioningServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -138,7 +138,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStartRunningServer() = runBlockingSimulation {
+    fun testStartRunningServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -153,7 +153,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStopProvisioningServer() = runBlockingSimulation {
+    fun testStopProvisioningServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -171,7 +171,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStopTerminatedServer() = runBlockingSimulation {
+    fun testStopTerminatedServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -185,7 +185,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStopDeletedServer() = runBlockingSimulation {
+    fun testStopDeletedServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -199,7 +199,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testStopRunningServer() = runBlockingSimulation {
+    fun testStopRunningServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>()
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -216,7 +216,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testDeleteProvisioningServer() = runBlockingSimulation {
+    fun testDeleteProvisioningServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>(relaxUnitFun = true)
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -235,7 +235,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testDeleteTerminatedServer() = runBlockingSimulation {
+    fun testDeleteTerminatedServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>(relaxUnitFun = true)
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -251,7 +251,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testDeleteDeletedServer() = runBlockingSimulation {
+    fun testDeleteDeletedServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>(relaxUnitFun = true)
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
@@ -265,7 +265,7 @@ class InternalServerTest {
     }
 
     @Test
-    fun testDeleteRunningServer() = runBlockingSimulation {
+    fun testDeleteRunningServer() = runSimulation {
         val service = mockk<ComputeServiceImpl>(relaxUnitFun = true)
         val uid = UUID.randomUUID()
         val flavor = mockFlavor()
diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
index a7993291..f63d4c6f 100644
--- a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
+++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt
@@ -41,9 +41,9 @@ import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.workload.SimTrace
 import org.opendc.simulator.compute.workload.SimTraceFragment
 import org.opendc.simulator.compute.workload.SimTraceWorkload
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.mux.FlowMultiplexerFactory
+import org.opendc.simulator.kotlin.runSimulation
 import java.time.Instant
 import java.util.*
 import kotlin.coroutines.resume
@@ -68,7 +68,7 @@ internal class SimHostTest {
      * Test overcommitting of resources by the hypervisor.
      */
     @Test
-    fun testOvercommitted() = runBlockingSimulation {
+    fun testOvercommitted() = runSimulation {
         val duration = 5 * 60L
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
@@ -152,7 +152,7 @@ internal class SimHostTest {
      * Test failure of the host.
      */
     @Test
-    fun testFailure() = runBlockingSimulation {
+    fun testFailure() = runSimulation {
         val duration = 5 * 60L
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt
index f240a25f..90f534e6 100644
--- a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt
+++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/failure/HostFaultInjectorTest.kt
@@ -29,7 +29,7 @@ import org.apache.commons.math3.distribution.LogNormalDistribution
 import org.apache.commons.math3.random.Well19937c
 import org.junit.jupiter.api.Test
 import org.opendc.compute.simulator.SimHost
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.time.Clock
 import java.time.Duration
 import kotlin.coroutines.CoroutineContext
@@ -43,7 +43,7 @@ internal class HostFaultInjectorTest {
      * Simple test case to test that nothing happens when the injector is not started.
      */
     @Test
-    fun testInjectorNotStarted() = runBlockingSimulation {
+    fun testInjectorNotStarted() = runSimulation {
         val host = mockk<SimHost>(relaxUnitFun = true)
 
         val injector = createSimpleInjector(coroutineContext, clock, setOf(host))
@@ -58,7 +58,7 @@ internal class HostFaultInjectorTest {
      * Simple test case to test a start stop fault where the machine is stopped and started after some time.
      */
     @Test
-    fun testInjectorStopsMachine() = runBlockingSimulation {
+    fun testInjectorStopsMachine() = runSimulation {
         val host = mockk<SimHost>(relaxUnitFun = true)
 
         val injector = createSimpleInjector(coroutineContext, clock, setOf(host))
@@ -77,7 +77,7 @@ internal class HostFaultInjectorTest {
      * Simple test case to test a start stop fault where multiple machines are stopped.
      */
     @Test
-    fun testInjectorStopsMultipleMachines() = runBlockingSimulation {
+    fun testInjectorStopsMultipleMachines() = runSimulation {
         val hosts = listOf<SimHost>(
             mockk(relaxUnitFun = true),
             mockk(relaxUnitFun = true)
diff --git a/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt b/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt
index f021e223..ef3deeb2 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt
@@ -32,7 +32,7 @@ import org.opendc.experiments.capelin.topology.clusterTopology
 import org.opendc.experiments.compute.*
 import org.opendc.experiments.compute.topology.HostSpec
 import org.opendc.experiments.provisioner.Provisioner
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import org.openjdk.jmh.annotations.*
 import java.io.File
 import java.util.*
@@ -60,7 +60,7 @@ class CapelinBenchmarks {
     }
 
     @Benchmark
-    fun benchmarkCapelin() = runBlockingSimulation {
+    fun benchmarkCapelin() = runSimulation {
         val serviceDomain = "compute.opendc.org"
 
         Provisioner(coroutineContext, clock, seed = 0).use { provisioner ->
diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt
index f1214b08..7461038d 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt
@@ -28,7 +28,7 @@ import org.opendc.experiments.capelin.topology.clusterTopology
 import org.opendc.experiments.compute.*
 import org.opendc.experiments.compute.export.parquet.ParquetComputeMonitor
 import org.opendc.experiments.provisioner.Provisioner
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.io.File
 import java.time.Duration
 import java.util.*
@@ -54,7 +54,7 @@ public class CapelinRunner(
     /**
      * Run a single [scenario] with the specified seed.
      */
-    fun runScenario(scenario: Scenario, seed: Long) = runBlockingSimulation {
+    fun runScenario(scenario: Scenario, seed: Long) = runSimulation {
         val serviceDomain = "compute.opendc.org"
         val topology = clusterTopology(File(envPath, "${scenario.topology.name}.txt"))
 
diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
index 9be2d522..238a5f87 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
@@ -39,7 +39,7 @@ import org.opendc.experiments.compute.telemetry.table.HostTableReader
 import org.opendc.experiments.compute.telemetry.table.ServiceTableReader
 import org.opendc.experiments.compute.topology.HostSpec
 import org.opendc.experiments.provisioner.Provisioner
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.io.File
 import java.time.Duration
 import java.util.*
@@ -80,7 +80,7 @@ class CapelinIntegrationTest {
      * Test a large simulation setup.
      */
     @Test
-    fun testLarge() = runBlockingSimulation {
+    fun testLarge() = runSimulation {
         val seed = 0L
         val workload = createTestWorkload(1.0, seed)
         val topology = createTopology()
@@ -124,7 +124,7 @@ class CapelinIntegrationTest {
      * Test a small simulation setup.
      */
     @Test
-    fun testSmall() = runBlockingSimulation {
+    fun testSmall() = runSimulation {
         val seed = 1L
         val workload = createTestWorkload(0.25, seed)
         val topology = createTopology("single")
@@ -164,7 +164,7 @@ class CapelinIntegrationTest {
      * Test a small simulation setup with interference.
      */
     @Test
-    fun testInterference() = runBlockingSimulation {
+    fun testInterference() = runSimulation {
         val seed = 0L
         val workload = createTestWorkload(1.0, seed)
         val topology = createTopology("single")
@@ -202,7 +202,7 @@ class CapelinIntegrationTest {
      * Test a small simulation setup with failures.
      */
     @Test
-    fun testFailures() = runBlockingSimulation {
+    fun testFailures() = runSimulation {
         val seed = 0L
         val topology = createTopology("single")
         val workload = createTestWorkload(0.25, seed)
diff --git a/opendc-experiments/opendc-experiments-faas/src/test/kotlin/org/opendc/experiments/faas/FaaSExperiment.kt b/opendc-experiments/opendc-experiments-faas/src/test/kotlin/org/opendc/experiments/faas/FaaSExperiment.kt
index 98328d3e..6f212825 100644
--- a/opendc-experiments/opendc-experiments-faas/src/test/kotlin/org/opendc/experiments/faas/FaaSExperiment.kt
+++ b/opendc-experiments/opendc-experiments-faas/src/test/kotlin/org/opendc/experiments/faas/FaaSExperiment.kt
@@ -34,7 +34,7 @@ import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.io.File
 import java.time.Duration
 
@@ -46,7 +46,7 @@ class FaaSExperiment {
      * Smoke test that simulates a small trace.
      */
     @Test
-    fun testSmoke() = runBlockingSimulation {
+    fun testSmoke() = runSimulation {
         val faasService = "faas.opendc.org"
 
         Provisioner(coroutineContext, clock, seed = 0L).use { provisioner ->
diff --git a/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/TensorFlowTest.kt b/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/TensorFlowTest.kt
index 328f1326..eee8b730 100644
--- a/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/TensorFlowTest.kt
+++ b/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/TensorFlowTest.kt
@@ -30,7 +30,7 @@ import org.opendc.experiments.tf20.distribute.MirroredStrategy
 import org.opendc.experiments.tf20.distribute.OneDeviceStrategy
 import org.opendc.experiments.tf20.util.MLEnvironmentReader
 import org.opendc.simulator.compute.power.LinearPowerModel
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -41,7 +41,7 @@ class TensorFlowTest {
      * Smoke test that tests the capabilities of the TensorFlow application model in OpenDC.
      */
     @Test
-    fun testSmokeAlexNet() = runBlockingSimulation {
+    fun testSmokeAlexNet() = runSimulation {
         val envInput = checkNotNull(TensorFlowTest::class.java.getResourceAsStream("/kth.json"))
         val def = MLEnvironmentReader().readEnvironment(envInput).first()
 
@@ -71,7 +71,7 @@ class TensorFlowTest {
      * Smoke test that tests the capabilities of the TensorFlow application model in OpenDC.
      */
     @Test
-    fun testSmokeVGG() = runBlockingSimulation {
+    fun testSmokeVGG() = runSimulation {
         val envInput = checkNotNull(TensorFlowTest::class.java.getResourceAsStream("/kth.json"))
         val def = MLEnvironmentReader().readEnvironment(envInput).first()
 
@@ -101,7 +101,7 @@ class TensorFlowTest {
      * Smoke test that tests the capabilities of the TensorFlow application model in OpenDC.
      */
     @Test
-    fun testSmokeDistribute() = runBlockingSimulation {
+    fun testSmokeDistribute() = runSimulation {
         val envInput = checkNotNull(TensorFlowTest::class.java.getResourceAsStream("/kth.json"))
         val def = MLEnvironmentReader().readEnvironment(envInput).first()
 
diff --git a/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/core/SimTFDeviceTest.kt b/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/core/SimTFDeviceTest.kt
index 051d5730..966ca5ef 100644
--- a/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/core/SimTFDeviceTest.kt
+++ b/opendc-experiments/opendc-experiments-tf20/src/test/kotlin/org/opendc/experiments/tf20/core/SimTFDeviceTest.kt
@@ -31,7 +31,7 @@ import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.power.LinearPowerModel
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -39,7 +39,7 @@ import java.util.*
  */
 internal class SimTFDeviceTest {
     @Test
-    fun testSmoke() = runBlockingSimulation {
+    fun testSmoke() = runSimulation {
         val puNode = ProcessingNode("NVIDIA", "Tesla V100", "unknown", 1)
         val pu = ProcessingUnit(puNode, 0, 960 * 1230.0)
         val memory = MemoryUnit("NVIDIA", "Tesla V100", 877.0, 32_000)
diff --git a/opendc-faas/opendc-faas-service/src/test/kotlin/org/opendc/faas/service/FaaSServiceTest.kt b/opendc-faas/opendc-faas-service/src/test/kotlin/org/opendc/faas/service/FaaSServiceTest.kt
index 560039c1..28234cf4 100644
--- a/opendc-faas/opendc-faas-service/src/test/kotlin/org/opendc/faas/service/FaaSServiceTest.kt
+++ b/opendc-faas/opendc-faas-service/src/test/kotlin/org/opendc/faas/service/FaaSServiceTest.kt
@@ -31,7 +31,7 @@ import org.opendc.faas.api.FaaSFunction
 import org.opendc.faas.service.deployer.FunctionDeployer
 import org.opendc.faas.service.deployer.FunctionInstance
 import org.opendc.faas.service.deployer.FunctionInstanceState
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -40,7 +40,7 @@ import java.util.*
 internal class FaaSServiceTest {
 
     @Test
-    fun testClientState() = runBlockingSimulation {
+    fun testClientState() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = assertDoesNotThrow { service.newClient() }
@@ -54,7 +54,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientInvokeUnknown() = runBlockingSimulation {
+    fun testClientInvokeUnknown() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -63,7 +63,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionCreation() = runBlockingSimulation {
+    fun testClientFunctionCreation() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -74,7 +74,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionQuery() = runBlockingSimulation {
+    fun testClientFunctionQuery() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -87,7 +87,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionFindById() = runBlockingSimulation {
+    fun testClientFunctionFindById() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -100,7 +100,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionFindByName() = runBlockingSimulation {
+    fun testClientFunctionFindByName() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -113,7 +113,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionDuplicateName() = runBlockingSimulation {
+    fun testClientFunctionDuplicateName() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -124,7 +124,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionDelete() = runBlockingSimulation {
+    fun testClientFunctionDelete() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -138,7 +138,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionCannotInvokeDeleted() = runBlockingSimulation {
+    fun testClientFunctionCannotInvokeDeleted() = runSimulation {
         val service = FaaSService(coroutineContext, clock, mockk(), mockk(), mockk())
 
         val client = service.newClient()
@@ -150,7 +150,7 @@ internal class FaaSServiceTest {
     }
 
     @Test
-    fun testClientFunctionInvoke() = runBlockingSimulation {
+    fun testClientFunctionInvoke() = runSimulation {
         val deployer = mockk<FunctionDeployer>()
         val service = FaaSService(coroutineContext, clock, deployer, mockk(), mockk(relaxUnitFun = true))
 
diff --git a/opendc-faas/opendc-faas-simulator/src/test/kotlin/org/opendc/faas/simulator/SimFaaSServiceTest.kt b/opendc-faas/opendc-faas-simulator/src/test/kotlin/org/opendc/faas/simulator/SimFaaSServiceTest.kt
index 5b730089..317eb0aa 100644
--- a/opendc-faas/opendc-faas-simulator/src/test/kotlin/org/opendc/faas/simulator/SimFaaSServiceTest.kt
+++ b/opendc-faas/opendc-faas-simulator/src/test/kotlin/org/opendc/faas/simulator/SimFaaSServiceTest.kt
@@ -42,7 +42,7 @@ import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimRuntimeWorkload
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import java.time.Duration
 import java.util.*
 
@@ -64,7 +64,7 @@ internal class SimFaaSServiceTest {
     }
 
     @Test
-    fun testSmoke() = runBlockingSimulation {
+    fun testSmoke() = runSimulation {
         val random = Random(0)
         val workload = spyk(object : SimFaaSWorkload, SimWorkload by SimRuntimeWorkload(1000) {
             override suspend fun invoke() {
diff --git a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
index 797d424e..b319a677 100644
--- a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
@@ -33,9 +33,9 @@ import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.workload.SimTrace
 import org.opendc.simulator.compute.workload.SimTraceWorkload
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.mux.FlowMultiplexerFactory
+import org.opendc.simulator.kotlin.runSimulation
 import org.openjdk.jmh.annotations.*
 import java.util.SplittableRandom
 import java.util.concurrent.ThreadLocalRandom
@@ -70,18 +70,18 @@ class SimMachineBenchmarks {
 
     @Benchmark
     fun benchmarkBareMetal() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val machine = SimBareMetalMachine(
                 engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            return@runBlockingSimulation machine.runWorkload(SimTraceWorkload(trace))
+            return@runSimulation machine.runWorkload(SimTraceWorkload(trace))
         }
     }
 
     @Benchmark
     fun benchmarkSpaceSharedHypervisor() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
             val hypervisor = SimHypervisor(engine, FlowMultiplexerFactory.forwardingMultiplexer(), SplittableRandom(1), null)
@@ -91,7 +91,7 @@ class SimMachineBenchmarks {
             val vm = hypervisor.newMachine(machineModel)
 
             try {
-                return@runBlockingSimulation vm.runWorkload(SimTraceWorkload(trace))
+                return@runSimulation vm.runWorkload(SimTraceWorkload(trace))
             } finally {
                 vm.cancel()
                 machine.cancel()
@@ -101,7 +101,7 @@ class SimMachineBenchmarks {
 
     @Benchmark
     fun benchmarkFairShareHypervisorSingle() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
             val hypervisor = SimHypervisor(engine, FlowMultiplexerFactory.maxMinMultiplexer(), SplittableRandom(1), null)
@@ -111,7 +111,7 @@ class SimMachineBenchmarks {
             val vm = hypervisor.newMachine(machineModel)
 
             try {
-                return@runBlockingSimulation vm.runWorkload(SimTraceWorkload(trace))
+                return@runSimulation vm.runWorkload(SimTraceWorkload(trace))
             } finally {
                 vm.cancel()
                 machine.cancel()
@@ -121,7 +121,7 @@ class SimMachineBenchmarks {
 
     @Benchmark
     fun benchmarkFairShareHypervisorDouble() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
             val hypervisor = SimHypervisor(engine, FlowMultiplexerFactory.maxMinMultiplexer(), SplittableRandom(1), null)
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
index 644eb497..b7af6803 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
@@ -33,9 +33,9 @@ import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.workload.SimFlopsWorkload
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.compute.workload.SimWorkloadLifecycle
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 import org.opendc.simulator.network.SimNetworkSink
 import org.opendc.simulator.power.SimPowerSource
 
@@ -58,7 +58,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testFlopsWorkload() = runBlockingSimulation {
+    fun testFlopsWorkload() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -72,7 +72,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testDualSocketMachine() = runBlockingSimulation {
+    fun testDualSocketMachine() = runSimulation {
         val cpuNode = machineModel.cpus[0].node
         val machineModel = MachineModel(
             cpus = List(cpuNode.coreCount * 2) { ProcessingUnit(cpuNode, it % 2, 1000.0) },
@@ -91,7 +91,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testPower() = runBlockingSimulation {
+    fun testPower() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             engine,
@@ -111,7 +111,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testCapacityClamp() = runBlockingSimulation {
+    fun testCapacityClamp() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -135,7 +135,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testMemory() = runBlockingSimulation {
+    fun testMemory() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -153,7 +153,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testMemoryUsage() = runBlockingSimulation {
+    fun testMemoryUsage() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -173,7 +173,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testNetUsage() = runBlockingSimulation {
+    fun testNetUsage() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             engine,
@@ -198,7 +198,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testDiskReadUsage() = runBlockingSimulation {
+    fun testDiskReadUsage() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             engine,
@@ -220,7 +220,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testDiskWriteUsage() = runBlockingSimulation {
+    fun testDiskWriteUsage() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             engine,
@@ -242,7 +242,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testCancellation() = runBlockingSimulation {
+    fun testCancellation() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -262,7 +262,7 @@ class SimMachineTest {
     }
 
     @Test
-    fun testConcurrentRuns() = runBlockingSimulation {
+    fun testConcurrentRuns() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt
index e5b509f0..5481cad2 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt
@@ -28,8 +28,8 @@ import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.compute.power.PowerDriver
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
+import org.opendc.simulator.kotlin.runSimulation
 import org.opendc.simulator.power.SimPowerSource
 
 /**
@@ -51,7 +51,7 @@ internal class SimPsuTest {
     }
 
     @Test
-    fun testPsuIdle() = runBlockingSimulation {
+    fun testPsuIdle() = runSimulation {
         val ratedOutputPower = 240.0
         val energyEfficiency = mapOf(0.0 to 1.0)
 
@@ -69,7 +69,7 @@ internal class SimPsuTest {
     }
 
     @Test
-    fun testPsuPowerLoss() = runBlockingSimulation {
+    fun testPsuPowerLoss() = runSimulation {
         val ratedOutputPower = 240.0
         // Efficiency of 80 Plus Titanium PSU
         val energyEfficiency = sortedMapOf(
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
index ddf8cf14..aae8d139 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
@@ -41,9 +41,9 @@ import org.opendc.simulator.compute.runWorkload
 import org.opendc.simulator.compute.workload.SimTrace
 import org.opendc.simulator.compute.workload.SimTraceFragment
 import org.opendc.simulator.compute.workload.SimTraceWorkload
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.mux.FlowMultiplexerFactory
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -65,7 +65,7 @@ internal class SimFairShareHypervisorTest {
      * Test overcommitting of resources via the hypervisor with a single VM.
      */
     @Test
-    fun testOvercommittedSingle() = runBlockingSimulation {
+    fun testOvercommittedSingle() = runSimulation {
         val duration = 5 * 60L
         val workloadA =
             SimTraceWorkload(
@@ -105,7 +105,7 @@ internal class SimFairShareHypervisorTest {
      * Test overcommitting of resources via the hypervisor with two VMs.
      */
     @Test
-    fun testOvercommittedDual() = runBlockingSimulation {
+    fun testOvercommittedDual() = runSimulation {
         val duration = 5 * 60L
         val workloadA =
             SimTraceWorkload(
@@ -158,7 +158,7 @@ internal class SimFairShareHypervisorTest {
     }
 
     @Test
-    fun testMultipleCPUs() = runBlockingSimulation {
+    fun testMultipleCPUs() = runSimulation {
         val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
         val model = MachineModel(
             cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
@@ -179,7 +179,7 @@ internal class SimFairShareHypervisorTest {
     }
 
     @Test
-    fun testInterference() = runBlockingSimulation {
+    fun testInterference() = runSimulation {
         val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
         val model = MachineModel(
             cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt
index df6755f1..664bb2da 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt
@@ -37,9 +37,9 @@ import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.runWorkload
 import org.opendc.simulator.compute.workload.*
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.mux.FlowMultiplexerFactory
+import org.opendc.simulator.kotlin.runSimulation
 import java.util.*
 
 /**
@@ -61,7 +61,7 @@ internal class SimSpaceSharedHypervisorTest {
      * Test a trace workload.
      */
     @Test
-    fun testTrace() = runBlockingSimulation {
+    fun testTrace() = runSimulation {
         val duration = 5 * 60L
         val workloadA =
             SimTraceWorkload(
@@ -92,7 +92,7 @@ internal class SimSpaceSharedHypervisorTest {
      * Test runtime workload on hypervisor.
      */
     @Test
-    fun testRuntimeWorkload() = runBlockingSimulation {
+    fun testRuntimeWorkload() = runSimulation {
         val duration = 5 * 60L * 1000
         val workload = SimRuntimeWorkload(duration)
         val engine = FlowEngine(coroutineContext, clock)
@@ -114,7 +114,7 @@ internal class SimSpaceSharedHypervisorTest {
      * Test FLOPs workload on hypervisor.
      */
     @Test
-    fun testFlopsWorkload() = runBlockingSimulation {
+    fun testFlopsWorkload() = runSimulation {
         val duration = 5 * 60L * 1000
         val workload = SimFlopsWorkload((duration * 3.2).toLong(), 1.0)
         val engine = FlowEngine(coroutineContext, clock)
@@ -134,7 +134,7 @@ internal class SimSpaceSharedHypervisorTest {
      * Test two workloads running sequentially.
      */
     @Test
-    fun testTwoWorkloads() = runBlockingSimulation {
+    fun testTwoWorkloads() = runSimulation {
         val duration = 5 * 60L * 1000
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
@@ -162,7 +162,7 @@ internal class SimSpaceSharedHypervisorTest {
      * Test concurrent workloads on the machine.
      */
     @Test
-    fun testConcurrentWorkloadFails() = runBlockingSimulation {
+    fun testConcurrentWorkloadFails() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
         val hypervisor = SimHypervisor(engine, FlowMultiplexerFactory.forwardingMultiplexer(), SplittableRandom(1), null)
@@ -184,7 +184,7 @@ internal class SimSpaceSharedHypervisorTest {
      * Test concurrent workloads on the machine.
      */
     @Test
-    fun testConcurrentWorkloadSucceeds() = runBlockingSimulation {
+    fun testConcurrentWorkloadSucceeds() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
         val hypervisor = SimHypervisor(engine, FlowMultiplexerFactory.forwardingMultiplexer(), SplittableRandom(1), null)
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
index aa91984a..70aea3f4 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
@@ -31,8 +31,8 @@ import org.opendc.simulator.compute.model.*
 import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.runWorkload
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [SimTraceWorkloadTest] class.
@@ -51,7 +51,7 @@ class SimTraceWorkloadTest {
     }
 
     @Test
-    fun testSmoke() = runBlockingSimulation {
+    fun testSmoke() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -74,7 +74,7 @@ class SimTraceWorkloadTest {
     }
 
     @Test
-    fun testOffset() = runBlockingSimulation {
+    fun testOffset() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -97,7 +97,7 @@ class SimTraceWorkloadTest {
     }
 
     @Test
-    fun testSkipFragment() = runBlockingSimulation {
+    fun testSkipFragment() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
@@ -121,7 +121,7 @@ class SimTraceWorkloadTest {
     }
 
     @Test
-    fun testZeroCores() = runBlockingSimulation {
+    fun testZeroCores() = runSimulation {
         val machine = SimBareMetalMachine(
             FlowEngine(coroutineContext, clock),
             machineModel,
diff --git a/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java
new file mode 100644
index 00000000..a70c1cda
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java
@@ -0,0 +1,246 @@
+/*
+ * Copyright (c) 2022 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;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.concurrent.Executor;
+
+/**
+ * A scheduler is used by simulations to manage execution of (future) tasks, providing a controllable (virtual) clock to
+ * skip over delays.
+ *
+ * <p>
+ * The scheduler can be queried to advance the time (via {@link #advanceBy}), run all the scheduled tasks advancing the
+ * virtual time as needed (via {@link #advanceUntilIdle}), or run the tasks that are scheduled to run as soon as
+ * possible but have not yet been dispatched (via {@link #runCurrent}). These methods execute the pending tasks using
+ * a single thread.
+ *
+ * <p>
+ * This class is not thread-safe and must not be used concurrently by multiple threads.
+ */
+public final class SimulationScheduler implements Executor {
+    /**
+     * The {@link TaskQueue} containing the pending tasks.
+     */
+    private final TaskQueue queue = new TaskQueue();
+
+    /**
+     * The current time of the scheduler in milliseconds since epoch.
+     */
+    private long currentTime;
+
+    /**
+     * A counter to establish total order on the events that happen at the same virtual time.
+     */
+    private int count = 0;
+
+    /**
+     * The {@link Clock} instance linked to this scheduler.
+     */
+    private final SimulationClock clock = new SimulationClock(this, ZoneId.systemDefault());
+
+    /**
+     * Construct a {@link SimulationScheduler} instance with the specified initial time.
+     *
+     * @param initialTimeMs The initial virtual time of the scheduler in milliseconds since epoch.
+     */
+    public SimulationScheduler(long initialTimeMs) {
+       this.currentTime = initialTimeMs;
+    }
+
+    /**
+     * Construct a {@link SimulationScheduler} instance with the initial time set to UNIX Epoch 0.
+     */
+    public SimulationScheduler() {
+        this(0);
+    }
+
+    /**
+     * Return the virtual clock associated with this dispatcher.
+     *
+     * @return A {@link Clock} tracking the virtual time of the dispatcher.
+     */
+    public Clock getClock() {
+        return clock;
+    }
+
+    /**
+     * Return the current virtual timestamp of the dispatcher (in milliseconds since epoch).
+     *
+     * @return A long value representing the virtual timestamp of the dispatcher in milliseconds since epoch.
+     */
+    public long getCurrentTime() {
+        return currentTime;
+    }
+
+    /**
+     * Schedule a <code>task</code> that executes after the specified <code>delayMs</code>.
+     *
+     * @param delayMs The time from now until the execution of the task (in milliseconds).
+     * @param task The task to execute after the delay.
+     * @return The identifier of the task that can be used together with the timestamp of the task to cancel it.
+     */
+    public int schedule(long delayMs, Runnable task) {
+        if (delayMs < 0) {
+            throw new IllegalArgumentException("Attempted scheduling an event earlier in time (delay " + delayMs + " ms)");
+        }
+
+        long target = currentTime + delayMs;
+        if (target < 0) {
+            target = Long.MAX_VALUE;
+        }
+
+        int id = count++;
+        queue.add(target, id, task);
+        return id;
+    }
+
+    /**
+     * Cancel a pending task.
+     *
+     * @param deadline The deadline of the task.
+     * @param id The identifier of the task (returned by {@link #schedule(long, Runnable)}).
+     * @return A boolean indicating whether a task was actually cancelled.
+     */
+    public boolean cancel(long deadline, int id) {
+        return queue.remove(deadline, id);
+    }
+
+    /**
+     * Run the enqueued tasks in the specified order, advancing the virtual time as needed until there are no more
+     * tasks in the queue of this scheduler.
+     */
+    public void advanceUntilIdle() {
+        final TaskQueue queue = this.queue;
+
+        while (true) {
+            long deadline = queue.peekDeadline();
+            Runnable task = queue.poll();
+
+            if (task == null) {
+                break;
+            }
+
+            currentTime = deadline;
+            task.run();
+        }
+    }
+
+    /**
+     * Move the virtual clock of this dispatcher forward by the specified amount, running the scheduled tasks in the
+     * meantime.
+     *
+     * @param delayMs The amount of time to move the virtual clock forward (in milliseconds).
+     * @throws IllegalStateException if passed a negative <code>delay</code>.
+     */
+    public void advanceBy(long delayMs) {
+        if (delayMs < 0) {
+            throw new IllegalArgumentException("Can not advance time by a negative delay: " + delayMs + " ms");
+        }
+
+        long target = currentTime + delayMs;
+        if (target < 0) {
+            target = Long.MAX_VALUE;
+        }
+
+        final TaskQueue queue = this.queue;
+        long deadline;
+
+        while ((deadline = queue.peekDeadline()) < target) {
+            Runnable task = queue.poll(); // Cannot be null since while condition is always false on an empty queue
+
+            task.run();
+            currentTime = deadline;
+        }
+
+        currentTime = target;
+    }
+
+    /**
+     * Execute the tasks that are scheduled to execute at this moment of virtual time.
+     */
+    public void runCurrent() {
+        final TaskQueue queue = this.queue;
+        long currentTime = this.currentTime;
+
+        while (queue.peekDeadline() == currentTime) {
+            Runnable task = queue.poll();
+
+            if (task == null) {
+                break;
+            }
+
+            task.run();
+        }
+    }
+
+    /**
+     * Schedule the specified command to run at this moment of virtual time.
+     *
+     * @param command The command to execute.
+     */
+    @Override
+    public void execute(Runnable command) {
+        schedule(0, command);
+    }
+
+    /**
+     * A {@link Clock} implementation for a {@link SimulationScheduler}.
+     */
+    private static class SimulationClock extends Clock {
+        private final SimulationScheduler scheduler;
+        private final ZoneId zone;
+
+        SimulationClock(SimulationScheduler scheduler, ZoneId zone) {
+            this.scheduler = scheduler;
+            this.zone = zone;
+        }
+
+        @Override
+        public ZoneId getZone() {
+            return zone;
+        }
+
+        @Override
+        public Clock withZone(ZoneId zoneId) {
+            return new SimulationClock(scheduler, zone);
+        }
+
+        @Override
+        public Instant instant() {
+            return Instant.ofEpochMilli(scheduler.currentTime);
+        }
+
+        @Override
+        public long millis() {
+            return scheduler.currentTime;
+        }
+
+        @Override
+        public String toString() {
+            return "SimulationClock[time=" + millis() + "ms]";
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/TaskQueue.java b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/TaskQueue.java
new file mode 100644
index 00000000..7d867b5d
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/TaskQueue.java
@@ -0,0 +1,265 @@
+/*
+ * Copyright (c) 2022 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;
+
+import java.util.Arrays;
+
+/**
+ * Specialized priority queue for pending tasks.
+ *
+ * <p>
+ * This class uses a specialized priority queue (as opposed to a generic {@link java.util.PriorityQueue}), which reduces
+ * unnecessary allocations in the simulator's hot path.
+ */
+final class TaskQueue {
+    /**
+     * The deadlines of the pending tasks.
+     */
+    private long[] deadlines;
+
+    /**
+     * The identifiers of the pending tasks. Identifiers are used to provide a total order for pending tasks in case
+     * the deadline of two tasks is the same.
+     */
+    private int[] ids;
+
+    /**
+     * The {@link Runnable}s representing the tasks that have been scheduled.
+     */
+    private Runnable[] tasks;
+
+    /**
+     * The number of elements in the priority queue.
+     */
+    private int size = 0;
+
+    /**
+     * Construct a {@link TaskQueue} with the specified initial capacity.
+     *
+     * @param initialCapacity The initial capacity of the queue.
+     */
+    public TaskQueue(int initialCapacity) {
+        this.deadlines = new long[initialCapacity];
+        this.ids = new int[initialCapacity];
+        this.tasks = new Runnable[initialCapacity];
+    }
+
+    /**
+     * Construct a {@link TaskQueue} with an initial capacity of 256 elements.
+     */
+    public TaskQueue() {
+        this(256);
+    }
+
+    /**
+     * Add a new task to this queue.
+     *
+     * @param deadline The deadline of the task.
+     * @param id       The identifier of the task.
+     * @param task     The {@link Runnable} representing the task to execute.
+     */
+    public void add(long deadline, int id, Runnable task) {
+        int i = size;
+        long[] deadlines = this.deadlines;
+
+        if (i >= deadlines.length) {
+            grow();
+
+            // Re-fetch the resized array
+            deadlines = this.deadlines;
+        }
+
+        siftUp(deadlines, ids, tasks, i, deadline, id, task);
+
+        size = i + 1;
+    }
+
+    /**
+     * Retrieve the next task to be executed.
+     *
+     * @return The head of the queue or <code>null</code> if the queue is empty.
+     */
+    public Runnable poll() {
+        final Runnable[] tasks = this.tasks;
+        final Runnable result = tasks[0];
+
+        if (result != null) {
+            int n = --size;
+
+            if (n > 0) {
+                long[] deadlines = this.deadlines;
+                int[] ids = this.ids;
+
+                siftDown(deadlines, ids, tasks, 0, n, deadlines[n], ids[n], tasks[n]);
+            }
+
+            // Clear the last element of the queue
+            tasks[n] = null;
+        }
+
+        return result;
+    }
+
+    /**
+     * Find the earliest deadline in the queue.
+     *
+     * @return The earliest deadline in the queue or {@link Long#MAX_VALUE} if the queue is empty.
+     */
+    public long peekDeadline() {
+        if (size == 0) {
+            return Long.MAX_VALUE;
+        }
+
+        return deadlines[0];
+    }
+
+    /**
+     * Remove the timer entry with the specified <code>deadline</code> and <code>id</code>.
+     */
+    public boolean remove(long deadline, int id) {
+        long[] deadlines = this.deadlines;
+        int[] ids = this.ids;
+
+        int size = this.size;
+        int i = -1;
+
+        for (int j = 0; j < size; j++) {
+            if (deadlines[j] == deadline && ids[j] == id) {
+                i = j;
+                break;
+            }
+        }
+
+        if (i < 0) {
+            return false;
+        }
+
+        Runnable[] tasks = this.tasks;
+        int s = size - 1;
+        this.size = s;
+
+        if (s == i) {
+            tasks[i] = null;
+        } else {
+            long movedDeadline = deadlines[s];
+            int movedId = ids[s];
+            Runnable movedTask = tasks[s];
+
+            tasks[s] = null;
+
+            siftDown(deadlines, ids, tasks, i, s, movedDeadline, movedId, movedTask);
+            if (tasks[i] == movedTask) {
+                siftUp(deadlines, ids, tasks, i, movedDeadline, movedId, movedTask);
+            }
+        }
+
+        return true;
+    }
+
+    /**
+     * Increases the capacity of the priority queue.
+     */
+    private void grow() {
+        int oldCapacity = deadlines.length;
+
+        // Double size if small; else grow by 50%
+        int newCapacity = oldCapacity + oldCapacity < 64 ? oldCapacity + 2 : oldCapacity >> 1;
+
+        deadlines = Arrays.copyOf(deadlines, newCapacity);
+        ids = Arrays.copyOf(ids, newCapacity);
+        tasks = Arrays.copyOf(tasks, newCapacity);
+    }
+
+    /**
+     * Sift up an entry in the heap.
+     */
+    private static void siftUp(long[] deadlines, int[] ids, Runnable[] tasks, int k, long deadline, int id, Runnable task) {
+        while (k > 0) {
+            int parent = (k - 1) >>> 1;
+            long parentDeadline = deadlines[parent];
+            int parentId = ids[parent];
+
+            if (compare(deadline, id, parentDeadline, parentId) >= 0) {
+                break;
+            }
+
+            deadlines[k] = parentDeadline;
+            ids[k] = parentId;
+            tasks[k] = tasks[parent];
+
+            k = parent;
+        }
+
+        deadlines[k] = deadline;
+        ids[k] = id;
+        tasks[k] = task;
+    }
+
+    /**
+     * Sift down an entry in the heap.
+     */
+    private static void siftDown(long[] deadlines, int[] ids, Runnable[] tasks, int k, int n, long deadline, int id, Runnable task) {
+        int half = n >>> 1;           // loop while a non-leaf
+
+        while (k < half) {
+            int child = (k << 1) + 1;  // assume left child is least
+
+            long childDeadline = deadlines[child];
+            int childId = ids[child];
+
+            int right = child + 1;
+            if (right < n) {
+                long rightDeadline = deadlines[right];
+                int rightId = ids[right];
+
+                if (compare(childDeadline, childId, rightDeadline, rightId) > 0) {
+                    child = right;
+                    childDeadline = rightDeadline;
+                    childId = rightId;
+                }
+            }
+
+            if (compare(deadline, id, childDeadline, childId) <= 0) {
+                break;
+            }
+
+            deadlines[k] = childDeadline;
+            ids[k] = childId;
+            tasks[k] = tasks[child];
+
+            k = child;
+        }
+
+        deadlines[k] = deadline;
+        ids[k] = id;
+        tasks[k] = task;
+    }
+
+    /**
+     * Helper method to compare two task entries.
+     */
+    private static int compare(long leftDeadline, int leftId, long rightDeadline, int rightId) {
+        int cmp = Long.compare(leftDeadline, rightDeadline);
+        return cmp == 0 ? Integer.compare(leftId, rightId) : cmp;
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt
deleted file mode 100644
index 908e902a..00000000
--- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * 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.core
-
-import kotlinx.coroutines.*
-import java.lang.Runnable
-import java.time.Clock
-import java.time.Instant
-import java.time.ZoneId
-import java.util.*
-import kotlin.coroutines.CoroutineContext
-
-/**
- * A [CoroutineDispatcher] that performs both immediate execution of coroutines on the main thread and uses a virtual
- * clock for time management.
- */
-@OptIn(InternalCoroutinesApi::class)
-public class SimulationCoroutineDispatcher : CoroutineDispatcher(), SimulationController, Delay {
-    /**
-     * Queue of ordered tasks to run.
-     */
-    private val queue = PriorityQueue<TimedRunnable>()
-
-    /**
-     * Global order counter.
-     */
-    private var _counter = 0L
-
-    /**
-     * The current virtual time of simulation
-     */
-    private var _clock = SimClock()
-
-    /**
-     * The virtual clock of this dispatcher.
-     */
-    override val clock: Clock = ClockAdapter(_clock)
-
-    override fun dispatch(context: CoroutineContext, block: Runnable) {
-        block.run()
-    }
-
-    override fun dispatchYield(context: CoroutineContext, block: Runnable) {
-        post(block)
-    }
-
-    @OptIn(ExperimentalCoroutinesApi::class)
-    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
-        postDelayed(CancellableContinuationRunnable(continuation) { resumeUndispatched(Unit) }, timeMillis)
-    }
-
-    override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
-        val node = postDelayed(block, timeMillis)
-        return object : DisposableHandle {
-            override fun dispose() {
-                queue.remove(node)
-            }
-        }
-    }
-
-    override fun toString(): String {
-        return "SimulationCoroutineDispatcher[time=${_clock.time}ms, queued=${queue.size}]"
-    }
-
-    private fun post(block: Runnable) =
-        queue.add(TimedRunnable(block, _counter++))
-
-    private fun postDelayed(block: Runnable, delayTime: Long) =
-        TimedRunnable(block, _counter++, safePlus(_clock.time, delayTime))
-            .also {
-                queue.add(it)
-            }
-
-    private fun safePlus(currentTime: Long, delayTime: Long): Long {
-        check(delayTime >= 0)
-        val result = currentTime + delayTime
-        if (result < currentTime) return Long.MAX_VALUE // clamp on overflow
-        return result
-    }
-
-    override fun advanceUntilIdle(): Long {
-        val queue = queue
-        val clock = _clock
-        val oldTime = clock.time
-
-        while (true) {
-            val current = queue.poll() ?: break
-
-            // If the scheduled time is 0 (immediate) use current virtual time
-            if (current.time != 0L) {
-                clock.time = current.time
-            }
-
-            current.run()
-        }
-
-        return clock.time - oldTime
-    }
-
-    /**
-     * A helper class that holds the time of the simulation.
-     */
-    private class SimClock(@JvmField var time: Long = 0)
-
-    /**
-     * A helper class to expose a [Clock] instance for this dispatcher.
-     */
-    private class ClockAdapter(private val clock: SimClock, private val zone: ZoneId = ZoneId.systemDefault()) : Clock() {
-        override fun getZone(): ZoneId = zone
-
-        override fun withZone(zone: ZoneId): Clock = ClockAdapter(clock, zone)
-
-        override fun instant(): Instant = Instant.ofEpochMilli(millis())
-
-        override fun millis(): Long = clock.time
-
-        override fun toString(): String = "SimulationCoroutineDispatcher.ClockAdapter[time=${clock.time}]"
-    }
-
-    /**
-     * This class exists to allow cleanup code to avoid throwing for cancelled continuations scheduled
-     * in the future.
-     */
-    private class CancellableContinuationRunnable<T>(
-        @JvmField val continuation: CancellableContinuation<T>,
-        private val block: CancellableContinuation<T>.() -> Unit
-    ) : Runnable {
-        override fun run() = continuation.block()
-    }
-
-    /**
-     * A Runnable for our event loop that represents a task to perform at a time.
-     */
-    private class TimedRunnable(
-        @JvmField val runnable: Runnable,
-        private val count: Long = 0,
-        @JvmField val time: Long = 0
-    ) : Comparable<TimedRunnable>, Runnable by runnable {
-        override fun compareTo(other: TimedRunnable) = if (time == other.time) {
-            count.compareTo(other.count)
-        } else {
-            time.compareTo(other.time)
-        }
-
-        override fun toString() = "TimedRunnable[time=$time, run=$runnable]"
-    }
-}
diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationBuilders.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationBuilders.kt
index 9b284c11..c4cc0171 100644
--- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationBuilders.kt
+++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationBuilders.kt
@@ -20,19 +20,46 @@
  * SOFTWARE.
  */
 
-package org.opendc.simulator.core
+package org.opendc.simulator.kotlin
 
 import kotlinx.coroutines.*
+import org.opendc.simulator.SimulationScheduler
 import kotlin.coroutines.ContinuationInterceptor
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
 
 /**
- * Executes a [body] inside an immediate execution dispatcher.
+ * Executes [body] as a simulation in a new coroutine.
+ *
+ * This function behaves similarly to [runBlocking], with the difference that the code that it runs will skip delays.
+ * This allows to use [delay] in without causing the simulation to take more time than necessary.
+ *
+ * ```
+ * @Test
+ * fun exampleSimulation() = runSimulation {
+ *     val deferred = async {
+ *         delay(1_000)
+ *         async {
+ *             delay(1_000)
+ *         }.await()
+ *     }
+ *
+ *     deferred.await() // result available immediately
+ * }
+ * ```
+ *
+ * The simulation is run in a single thread, unless other [CoroutineDispatcher] are used for child coroutines.
+ * Because of this, child coroutines are not executed in parallel to [body].
+ * In order for the spawned-off asynchronous code to actually be executed, one must either [yield] or suspend the
+ * body some other way, or use commands that control scheduling (see [SimulationScheduler]).
  */
 @OptIn(ExperimentalCoroutinesApi::class)
-public fun runBlockingSimulation(context: CoroutineContext = EmptyCoroutineContext, body: suspend SimulationCoroutineScope.() -> Unit) {
-    val (safeContext, dispatcher) = context.checkArguments()
+public fun runSimulation(
+    context: CoroutineContext = EmptyCoroutineContext,
+    scheduler: SimulationScheduler = SimulationScheduler(),
+    body: suspend SimulationCoroutineScope.() -> Unit
+) {
+    val (safeContext, dispatcher) = context.checkArguments(scheduler)
     val startingJobs = safeContext.activeJobs()
     val scope = SimulationCoroutineScope(safeContext)
     val deferred = scope.async {
@@ -49,21 +76,21 @@ public fun runBlockingSimulation(context: CoroutineContext = EmptyCoroutineConte
 }
 
 /**
- * Convenience method for calling [runBlockingSimulation] on an existing [SimulationCoroutineScope].
+ * Convenience method for calling [runSimulation] on an existing [SimulationCoroutineScope].
  */
-public fun SimulationCoroutineScope.runBlockingSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit =
-    runBlockingSimulation(coroutineContext, block)
+public fun SimulationCoroutineScope.runSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit =
+    runSimulation(coroutineContext, scheduler, block)
 
 /**
- * Convenience method for calling [runBlockingSimulation] on an existing [SimulationCoroutineDispatcher].
+ * Convenience method for calling [runSimulation] on an existing [SimulationCoroutineDispatcher].
  */
-public fun SimulationCoroutineDispatcher.runBlockingSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit =
-    runBlockingSimulation(this, block)
+public fun SimulationCoroutineDispatcher.runSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit =
+    runSimulation(this, scheduler, block)
 
-private fun CoroutineContext.checkArguments(): Pair<CoroutineContext, SimulationController> {
+private fun CoroutineContext.checkArguments(scheduler: SimulationScheduler): Pair<CoroutineContext, SimulationController> {
     val dispatcher = get(ContinuationInterceptor).run {
         this?.let { require(this is SimulationController) { "Dispatcher must implement SimulationController: $this" } }
-        this ?: SimulationCoroutineDispatcher()
+        this ?: SimulationCoroutineDispatcher(scheduler)
     }
 
     val job = get(Job) ?: SupervisorJob()
diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationController.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationController.kt
index 2b670b91..f96b2326 100644
--- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationController.kt
+++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationController.kt
@@ -20,9 +20,10 @@
  * SOFTWARE.
  */
 
-package org.opendc.simulator.core
+package org.opendc.simulator.kotlin
 
 import kotlinx.coroutines.CoroutineDispatcher
+import org.opendc.simulator.SimulationScheduler
 import java.time.Clock
 
 /**
@@ -35,6 +36,11 @@ public interface SimulationController {
     public val clock: Clock
 
     /**
+     * The [SimulationScheduler] driving the simulation.
+     */
+    public val scheduler: SimulationScheduler
+
+    /**
      * Immediately execute all pending tasks and advance the virtual clock-time to the last delay.
      *
      * If new tasks are scheduled due to advancing virtual time, they will be executed before `advanceUntilIdle`
diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt
new file mode 100644
index 00000000..21ad1a86
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt
@@ -0,0 +1,94 @@
+/*
+ * 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.kotlin
+
+import kotlinx.coroutines.*
+import org.opendc.simulator.SimulationScheduler
+import java.lang.Runnable
+import java.time.Clock
+import java.util.*
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * A [CoroutineDispatcher] that performs both immediate execution of coroutines on the main thread and uses a virtual
+ * clock for time management.
+ *
+ * @param scheduler The [SimulationScheduler] used to manage the execution of future tasks.
+ */
+@OptIn(InternalCoroutinesApi::class)
+public class SimulationCoroutineDispatcher(
+    override val scheduler: SimulationScheduler = SimulationScheduler()
+) : CoroutineDispatcher(), SimulationController, Delay {
+    /**
+     * The virtual clock of this dispatcher.
+     */
+    override val clock: Clock = scheduler.clock
+
+    override fun dispatch(context: CoroutineContext, block: Runnable) {
+        block.run()
+    }
+
+    override fun dispatchYield(context: CoroutineContext, block: Runnable) {
+        scheduler.execute(block)
+    }
+
+    @OptIn(ExperimentalCoroutinesApi::class)
+    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
+        scheduler.schedule(timeMillis, CancellableContinuationRunnable(continuation) { resumeUndispatched(Unit) })
+    }
+
+    override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
+        return object : DisposableHandle {
+            private val deadline = (scheduler.currentTime + timeMillis).let { if (it >= 0) it else Long.MAX_VALUE }
+            private val id = scheduler.schedule(timeMillis, block)
+
+            override fun dispose() {
+                scheduler.cancel(deadline, id)
+            }
+        }
+    }
+
+    override fun toString(): String {
+        return "SimulationCoroutineDispatcher[time=${scheduler.currentTime}ms]"
+    }
+
+    override fun advanceUntilIdle(): Long {
+        val scheduler = scheduler
+        val oldTime = scheduler.currentTime
+
+        scheduler.advanceUntilIdle()
+
+        return scheduler.currentTime - oldTime
+    }
+
+    /**
+     * This class exists to allow cleanup code to avoid throwing for cancelled continuations scheduled
+     * in the future.
+     */
+    private class CancellableContinuationRunnable<T>(
+        @JvmField val continuation: CancellableContinuation<T>,
+        private val block: CancellableContinuation<T>.() -> Unit
+    ) : Runnable {
+        override fun run() = continuation.block()
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineScope.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineScope.kt
index 1da7f0fa..6be8e67a 100644
--- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineScope.kt
+++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineScope.kt
@@ -20,16 +20,17 @@
  * SOFTWARE.
  */
 
-package org.opendc.simulator.core
+package org.opendc.simulator.kotlin
 
 import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
+import org.opendc.simulator.SimulationScheduler
 import kotlin.coroutines.ContinuationInterceptor
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
 
 /**
- *  A scope which provides detailed control over the execution of coroutines for simulations.
+ * A scope which provides detailed control over the execution of coroutines for simulations.
  */
 public interface SimulationCoroutineScope : CoroutineScope, SimulationController
 
@@ -46,9 +47,12 @@ private class SimulationCoroutineScopeImpl(
  * scope adds [SimulationCoroutineDispatcher] automatically.
  */
 @Suppress("FunctionName")
-public fun SimulationCoroutineScope(context: CoroutineContext = EmptyCoroutineContext): SimulationCoroutineScope {
+public fun SimulationCoroutineScope(
+    context: CoroutineContext = EmptyCoroutineContext,
+    scheduler: SimulationScheduler = SimulationScheduler()
+): SimulationCoroutineScope {
     var safeContext = context
-    if (context[ContinuationInterceptor] == null) safeContext += SimulationCoroutineDispatcher()
+    if (context[ContinuationInterceptor] == null) safeContext += SimulationCoroutineDispatcher(scheduler)
     return SimulationCoroutineScopeImpl(safeContext)
 }
 
diff --git a/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt
new file mode 100644
index 00000000..eca3b582
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt
@@ -0,0 +1,106 @@
+/*
+ * Copyright (c) 2022 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
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import java.time.Instant
+
+/**
+ * Test suite for the [SimulationScheduler] class.
+ */
+class SimulationSchedulerTest {
+    /**
+     * Test the basic functionality of [SimulationScheduler.runCurrent].
+     */
+    @Test
+    fun testRunCurrent() {
+        val scheduler = SimulationScheduler()
+        var count = 0
+
+        scheduler.schedule(1) { count += 1 }
+        scheduler.schedule(2) { count += 1 }
+
+        scheduler.advanceBy(1)
+        assertEquals(0, count)
+        scheduler.runCurrent()
+        assertEquals(1, count)
+        scheduler.advanceBy(1)
+        assertEquals(1, count)
+        scheduler.runCurrent()
+        assertEquals(2, count)
+        assertEquals(2, scheduler.currentTime)
+
+        scheduler.advanceBy(Long.MAX_VALUE)
+        scheduler.runCurrent()
+        assertEquals(Long.MAX_VALUE, scheduler.currentTime)
+    }
+
+    /**
+     * Test the clock of the [SimulationScheduler].
+     */
+    @Test
+    fun testClock() {
+        val scheduler = SimulationScheduler()
+        var count = 0
+
+        scheduler.schedule(1) { count += 1 }
+        scheduler.schedule(2) { count += 1 }
+
+        scheduler.advanceBy(2)
+        assertEquals(2, scheduler.currentTime)
+        assertEquals(2, scheduler.clock.millis())
+        assertEquals(Instant.ofEpochMilli(2), scheduler.clock.instant())
+    }
+
+    /**
+     * Test large delays.
+     */
+    @Test
+    fun testAdvanceByLargeDelays() {
+        val scheduler = SimulationScheduler()
+        var count = 0
+
+        scheduler.schedule(1) { count += 1 }
+
+        scheduler.advanceBy(10)
+
+        scheduler.schedule(Long.MAX_VALUE) { count += 1 }
+        scheduler.schedule(100_000_000) { count += 1 }
+
+        scheduler.advanceUntilIdle()
+        assertEquals(3, count)
+    }
+
+    /**
+     * Test negative delays.
+     */
+    @Test
+    fun testNegativeDelays() {
+        val scheduler = SimulationScheduler()
+
+        assertThrows<IllegalArgumentException> { scheduler.schedule(-100) { } }
+        assertThrows<IllegalArgumentException> { scheduler.advanceBy(-100) }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/TaskQueueTest.kt b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/TaskQueueTest.kt
new file mode 100644
index 00000000..a4d779cb
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/TaskQueueTest.kt
@@ -0,0 +1,231 @@
+/*
+ * Copyright (c) 2022 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
+
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+
+/**
+ * Test suite for the [TaskQueue] class.
+ */
+class TaskQueueTest {
+    private lateinit var queue: TaskQueue
+
+    @BeforeEach
+    fun setUp() {
+        queue = TaskQueue(3)
+    }
+
+    /**
+     * Test whether a call to [TaskQueue.poll] returns `null` for an empty queue.
+     */
+    @Test
+    fun testPollEmpty() {
+        assertAll(
+            { assertEquals(Long.MAX_VALUE, queue.peekDeadline()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test whether a call to [TaskQueue.poll] returns the proper value for a queue with a single entry.
+     */
+    @Test
+    fun testSingleEntry() {
+        val entry = Runnable {}
+
+        queue.add(100, 1, entry)
+
+        assertAll(
+            { assertEquals(100, queue.peekDeadline()) },
+            { assertEquals(entry, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test whether [TaskQueue.poll] returns values in the queue in the proper order.
+     */
+    @Test
+    fun testMultipleEntries() {
+        val entryA = Runnable {}
+        queue.add(100, 1, entryA)
+
+        val entryB = Runnable {}
+        queue.add(48, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(58, 1, entryC)
+
+        assertAll(
+            { assertEquals(48, queue.peekDeadline()) },
+            { assertEquals(entryB, queue.poll()) },
+            { assertEquals(entryC, queue.poll()) },
+            { assertEquals(entryA, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test whether [TaskQueue.poll] returns values in the queue in the proper order with duplicates.
+     */
+    @Test
+    fun testMultipleEntriesDuplicate() {
+        val entryA = Runnable {}
+        queue.add(48, 0, entryA)
+
+        val entryB = Runnable {}
+        queue.add(48, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(48, 2, entryC)
+
+        assertAll(
+            { assertEquals(48, queue.peekDeadline()) },
+            { assertEquals(entryA, queue.poll()) },
+            { assertEquals(entryB, queue.poll()) },
+            { assertEquals(entryC, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test that the queue is properly resized when the number of entries exceed the capacity.
+     */
+    @Test
+    fun testResize() {
+        val entryA = Runnable {}
+        queue.add(100, 1, entryA)
+
+        val entryB = Runnable {}
+        queue.add(20, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(58, 1, entryC)
+
+        val entryD = Runnable {}
+        queue.add(38, 1, entryD)
+
+        assertAll(
+            { assertEquals(20, queue.peekDeadline()) },
+            { assertEquals(entryB, queue.poll()) },
+            { assertEquals(entryD, queue.poll()) },
+            { assertEquals(entryC, queue.poll()) },
+            { assertEquals(entryA, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test that we can remove an entry from the end of the queue.
+     */
+    @Test
+    fun testRemoveEntryTail() {
+        val entryA = Runnable {}
+        queue.add(100, 1, entryA)
+
+        val entryB = Runnable {}
+        queue.add(20, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(58, 1, entryC)
+
+        queue.remove(100, 1)
+
+        assertAll(
+            { assertEquals(20, queue.peekDeadline()) },
+            { assertEquals(entryB, queue.poll()) },
+            { assertEquals(entryC, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test that we can remove an entry from the head of the queue.
+     */
+    @Test
+    fun testRemoveEntryHead() {
+        val entryA = Runnable {}
+        queue.add(100, 1, entryA)
+
+        val entryB = Runnable {}
+        queue.add(20, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(58, 1, entryC)
+
+        queue.remove(20, 1)
+
+        assertAll(
+            { assertEquals(58, queue.peekDeadline()) },
+            { assertEquals(entryC, queue.poll()) },
+            { assertEquals(entryA, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test that we can remove an entry from the middle of a queue.
+     */
+    @Test
+    fun testRemoveEntryMiddle() {
+        val entryA = Runnable {}
+        queue.add(100, 1, entryA)
+
+        val entryB = Runnable {}
+        queue.add(20, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(58, 1, entryC)
+
+        queue.remove(58, 1)
+
+        assertAll(
+            { assertEquals(20, queue.peekDeadline()) },
+            { assertEquals(entryB, queue.poll()) },
+            { assertEquals(entryA, queue.poll()) },
+            { assertNull(queue.poll()) },
+        )
+    }
+
+    /**
+     * Test that we can "remove" an unknown entry without error.
+     */
+    @Test
+    fun testRemoveUnknown() {
+        val entryA = Runnable {}
+        queue.add(100, 1, entryA)
+
+        val entryB = Runnable {}
+        queue.add(20, 1, entryB)
+
+        val entryC = Runnable {}
+        queue.add(58, 1, entryC)
+
+        assertAll(
+            { assertFalse(queue.remove(10, 1)) },
+            { assertFalse(queue.remove(58, 2)) }
+        )
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt
index aabd2220..86fbe8e4 100644
--- a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt
@@ -24,10 +24,10 @@ package org.opendc.simulator.flow
 
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.launch
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.mux.ForwardingFlowMultiplexer
 import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
 import org.opendc.simulator.flow.source.TraceFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeUnit
@@ -49,27 +49,27 @@ class FlowBenchmarks {
 
     @Benchmark
     fun benchmarkSink() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val provider = FlowSink(engine, 4200.0)
-            return@runBlockingSimulation provider.consume(TraceFlowSource(trace))
+            return@runSimulation provider.consume(TraceFlowSource(trace))
         }
     }
 
     @Benchmark
     fun benchmarkForward() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val provider = FlowSink(engine, 4200.0)
             val forwarder = FlowForwarder(engine)
             provider.startConsumer(forwarder)
-            return@runBlockingSimulation forwarder.consume(TraceFlowSource(trace))
+            return@runSimulation forwarder.consume(TraceFlowSource(trace))
         }
     }
 
     @Benchmark
     fun benchmarkMuxMaxMinSingleSource() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val switch = MaxMinFlowMultiplexer(engine)
 
@@ -77,13 +77,13 @@ class FlowBenchmarks {
             FlowSink(engine, 3000.0).startConsumer(switch.newOutput())
 
             val provider = switch.newInput()
-            return@runBlockingSimulation provider.consume(TraceFlowSource(trace))
+            return@runSimulation provider.consume(TraceFlowSource(trace))
         }
     }
 
     @Benchmark
     fun benchmarkMuxMaxMinTripleSource() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val switch = MaxMinFlowMultiplexer(engine)
 
@@ -101,7 +101,7 @@ class FlowBenchmarks {
 
     @Benchmark
     fun benchmarkMuxExclusiveSingleSource() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val switch = ForwardingFlowMultiplexer(engine)
 
@@ -109,13 +109,13 @@ class FlowBenchmarks {
             FlowSink(engine, 3000.0).startConsumer(switch.newOutput())
 
             val provider = switch.newInput()
-            return@runBlockingSimulation provider.consume(TraceFlowSource(trace))
+            return@runSimulation provider.consume(TraceFlowSource(trace))
         }
     }
 
     @Benchmark
     fun benchmarkMuxExclusiveTripleSource() {
-        return runBlockingSimulation {
+        return runSimulation {
             val engine = FlowEngine(coroutineContext, clock)
             val switch = ForwardingFlowMultiplexer(engine)
 
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt
index e7b25554..d782d036 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt
@@ -24,16 +24,16 @@ package org.opendc.simulator.flow
 
 import io.mockk.*
 import org.junit.jupiter.api.*
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.internal.FlowConsumerContextImpl
 import org.opendc.simulator.flow.internal.FlowEngineImpl
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * A test suite for the [FlowConsumerContextImpl] class.
  */
 class FlowConsumerContextTest {
     @Test
-    fun testFlushWithoutCommand() = runBlockingSimulation {
+    fun testFlushWithoutCommand() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val consumer = object : FlowSource {
             override fun onPull(conn: FlowConnection, now: Long): Long {
@@ -54,7 +54,7 @@ class FlowConsumerContextTest {
     }
 
     @Test
-    fun testDoubleStart() = runBlockingSimulation {
+    fun testDoubleStart() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val consumer = object : FlowSource {
             override fun onPull(conn: FlowConnection, now: Long): Long {
@@ -79,7 +79,7 @@ class FlowConsumerContextTest {
     }
 
     @Test
-    fun testIdempotentCapacityChange() = runBlockingSimulation {
+    fun testIdempotentCapacityChange() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val consumer = spyk(object : FlowSource {
             override fun onPull(conn: FlowConnection, now: Long): Long {
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt
index 8b090593..2025dd52 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt
@@ -28,16 +28,16 @@ import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.internal.FlowEngineImpl
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * A test suite for the [FlowForwarder] class.
  */
 internal class FlowForwarderTest {
     @Test
-    fun testCancelImmediately() = runBlockingSimulation {
+    fun testCancelImmediately() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 2000.0)
@@ -56,7 +56,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testCancel() = runBlockingSimulation {
+    fun testCancel() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 2000.0)
@@ -83,7 +83,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testState() = runBlockingSimulation {
+    fun testState() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val consumer = object : FlowSource {
@@ -108,7 +108,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testCancelPendingDelegate() = runBlockingSimulation {
+    fun testCancelPendingDelegate() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
 
@@ -126,7 +126,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testCancelStartedDelegate() = runBlockingSimulation {
+    fun testCancelStartedDelegate() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 2000.0)
@@ -144,7 +144,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testCancelPropagation() = runBlockingSimulation {
+    fun testCancelPropagation() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 2000.0)
@@ -162,7 +162,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testExitPropagation() = runBlockingSimulation {
+    fun testExitPropagation() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine, isCoupled = true)
         val source = FlowSink(engine, 2000.0)
@@ -183,7 +183,7 @@ internal class FlowForwarderTest {
 
     @Test
     @Disabled // Due to Kotlin bug: https://github.com/mockk/mockk/issues/368
-    fun testAdjustCapacity() = runBlockingSimulation {
+    fun testAdjustCapacity() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val sink = FlowSink(engine, 1.0)
@@ -202,7 +202,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testCounters() = runBlockingSimulation {
+    fun testCounters() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 1.0)
@@ -224,7 +224,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testCoupledExit() = runBlockingSimulation {
+    fun testCoupledExit() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine, isCoupled = true)
         val source = FlowSink(engine, 2000.0)
@@ -239,7 +239,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testPullFailureCoupled() = runBlockingSimulation {
+    fun testPullFailureCoupled() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine, isCoupled = true)
         val source = FlowSink(engine, 2000.0)
@@ -262,7 +262,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testStartFailure() = runBlockingSimulation {
+    fun testStartFailure() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 2000.0)
@@ -290,7 +290,7 @@ internal class FlowForwarderTest {
     }
 
     @Test
-    fun testConvergeFailure() = runBlockingSimulation {
+    fun testConvergeFailure() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val forwarder = FlowForwarder(engine)
         val source = FlowSink(engine, 2000.0)
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt
index 726ddbf7..22a84edb 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt
@@ -27,17 +27,17 @@ import io.mockk.verify
 import kotlinx.coroutines.*
 import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.assertEquals
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.internal.FlowEngineImpl
 import org.opendc.simulator.flow.source.FixedFlowSource
 import org.opendc.simulator.flow.source.FlowSourceRateAdapter
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * A test suite for the [FlowSink] class.
  */
 internal class FlowSinkTest {
     @Test
-    fun testSpeed() = runBlockingSimulation {
+    fun testSpeed() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -53,7 +53,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testAdjustCapacity() = runBlockingSimulation {
+    fun testAdjustCapacity() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val provider = FlowSink(engine, 1.0)
 
@@ -69,7 +69,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testSpeedLimit() = runBlockingSimulation {
+    fun testSpeedLimit() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -89,7 +89,7 @@ internal class FlowSinkTest {
      * [FlowSource.onPull].
      */
     @Test
-    fun testIntermediateInterrupt() = runBlockingSimulation {
+    fun testIntermediateInterrupt() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -109,7 +109,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testInterrupt() = runBlockingSimulation {
+    fun testInterrupt() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -144,7 +144,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testFailure() = runBlockingSimulation {
+    fun testFailure() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -165,7 +165,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testExceptionPropagationOnNext() = runBlockingSimulation {
+    fun testExceptionPropagationOnNext() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -190,7 +190,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testConcurrentConsumption() = runBlockingSimulation {
+    fun testConcurrentConsumption() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -206,7 +206,7 @@ internal class FlowSinkTest {
     }
 
     @Test
-    fun testCancelDuringConsumption() = runBlockingSimulation {
+    fun testCancelDuringConsumption() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
         val capacity = 4200.0
         val provider = FlowSink(engine, capacity)
@@ -225,7 +225,7 @@ internal class FlowSinkTest {
     @Test
     fun testInfiniteSleep() {
         assertThrows<IllegalStateException> {
-            runBlockingSimulation {
+            runSimulation {
                 val engine = FlowEngineImpl(coroutineContext, clock)
                 val capacity = 4200.0
                 val provider = FlowSink(engine, capacity)
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt
index ef15f711..cfd2bdf0 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt
@@ -27,12 +27,12 @@ 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.core.runBlockingSimulation
 import org.opendc.simulator.flow.*
 import org.opendc.simulator.flow.internal.FlowEngineImpl
 import org.opendc.simulator.flow.source.FixedFlowSource
 import org.opendc.simulator.flow.source.FlowSourceRateAdapter
 import org.opendc.simulator.flow.source.TraceFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [ForwardingFlowMultiplexer] class.
@@ -42,7 +42,7 @@ internal class ForwardingFlowMultiplexerTest {
      * Test a trace workload.
      */
     @Test
-    fun testTrace() = runBlockingSimulation {
+    fun testTrace() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
 
         val speed = mutableListOf<Double>()
@@ -79,7 +79,7 @@ internal class ForwardingFlowMultiplexerTest {
      * Test runtime workload on hypervisor.
      */
     @Test
-    fun testRuntimeWorkload() = runBlockingSimulation {
+    fun testRuntimeWorkload() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
@@ -101,7 +101,7 @@ internal class ForwardingFlowMultiplexerTest {
      * Test two workloads running sequentially.
      */
     @Test
-    fun testTwoWorkloads() = runBlockingSimulation {
+    fun testTwoWorkloads() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
 
         val duration = 5 * 60L * 1000
@@ -140,7 +140,7 @@ internal class ForwardingFlowMultiplexerTest {
      * Test concurrent workloads on the machine.
      */
     @Test
-    fun testConcurrentWorkloadFails() = runBlockingSimulation {
+    fun testConcurrentWorkloadFails() = runSimulation {
         val engine = FlowEngineImpl(coroutineContext, clock)
 
         val switch = ForwardingFlowMultiplexer(engine)
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt
index 6e2cdb98..4e242292 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt
@@ -27,19 +27,19 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.yield
 import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.assertEquals
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowSink
 import org.opendc.simulator.flow.consume
 import org.opendc.simulator.flow.internal.FlowEngineImpl
 import org.opendc.simulator.flow.source.FixedFlowSource
 import org.opendc.simulator.flow.source.TraceFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [FlowMultiplexer] implementations
  */
 internal class MaxMinFlowMultiplexerTest {
     @Test
-    fun testSmoke() = runBlockingSimulation {
+    fun testSmoke() = runSimulation {
         val scheduler = FlowEngineImpl(coroutineContext, clock)
         val switch = MaxMinFlowMultiplexer(scheduler)
 
@@ -61,7 +61,7 @@ internal class MaxMinFlowMultiplexerTest {
      * Test overcommitting of resources via the hypervisor with a single VM.
      */
     @Test
-    fun testOvercommittedSingle() = runBlockingSimulation {
+    fun testOvercommittedSingle() = runSimulation {
         val scheduler = FlowEngineImpl(coroutineContext, clock)
 
         val duration = 5 * 60L
@@ -99,7 +99,7 @@ internal class MaxMinFlowMultiplexerTest {
      * Test overcommitting of resources via the hypervisor with two VMs.
      */
     @Test
-    fun testOvercommittedDual() = runBlockingSimulation {
+    fun testOvercommittedDual() = runSimulation {
         val scheduler = FlowEngineImpl(coroutineContext, clock)
 
         val duration = 5 * 60L
diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt
index 8396d346..552579ff 100644
--- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt
@@ -24,17 +24,17 @@ package org.opendc.simulator.flow.source
 
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowSink
 import org.opendc.simulator.flow.consume
 import org.opendc.simulator.flow.internal.FlowEngineImpl
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * A test suite for the [FixedFlowSource] class.
  */
 internal class FixedFlowSourceTest {
     @Test
-    fun testSmoke() = runBlockingSimulation {
+    fun testSmoke() = runSimulation {
         val scheduler = FlowEngineImpl(coroutineContext, clock)
         val provider = FlowSink(scheduler, 1.0)
 
@@ -45,7 +45,7 @@ internal class FixedFlowSourceTest {
     }
 
     @Test
-    fun testUtilization() = runBlockingSimulation {
+    fun testUtilization() = runSimulation {
         val scheduler = FlowEngineImpl(coroutineContext, clock)
         val provider = FlowSink(scheduler, 1.0)
 
diff --git a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt
index 2e6983c8..944c4d6a 100644
--- a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt
+++ b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt
@@ -30,16 +30,16 @@ import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertDoesNotThrow
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.*
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [SimNetworkSink] class.
  */
 class SimNetworkSinkTest {
     @Test
-    fun testInitialState() = runBlockingSimulation {
+    fun testInitialState() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
 
@@ -49,7 +49,7 @@ class SimNetworkSinkTest {
     }
 
     @Test
-    fun testDisconnectIdempotent() = runBlockingSimulation {
+    fun testDisconnectIdempotent() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
 
@@ -58,7 +58,7 @@ class SimNetworkSinkTest {
     }
 
     @Test
-    fun testConnectCircular() = runBlockingSimulation {
+    fun testConnectCircular() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
 
@@ -68,7 +68,7 @@ class SimNetworkSinkTest {
     }
 
     @Test
-    fun testConnectAlreadyConnectedTarget() = runBlockingSimulation {
+    fun testConnectAlreadyConnectedTarget() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
         val source = mockk<SimNetworkPort>(relaxUnitFun = true)
@@ -80,7 +80,7 @@ class SimNetworkSinkTest {
     }
 
     @Test
-    fun testConnectAlreadyConnected() = runBlockingSimulation {
+    fun testConnectAlreadyConnected() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
         val source1 = Source(engine)
@@ -96,7 +96,7 @@ class SimNetworkSinkTest {
     }
 
     @Test
-    fun testConnect() = runBlockingSimulation {
+    fun testConnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
         val source = spyk(Source(engine))
@@ -112,7 +112,7 @@ class SimNetworkSinkTest {
     }
 
     @Test
-    fun testDisconnect() = runBlockingSimulation {
+    fun testDisconnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
         val source = spyk(Source(engine))
diff --git a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt
index 62e54ffb..ff6cb66e 100644
--- a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt
+++ b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt
@@ -27,16 +27,16 @@ import io.mockk.verify
 import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.*
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [SimNetworkSwitchVirtual] class.
  */
 class SimNetworkSwitchVirtualTest {
     @Test
-    fun testConnect() = runBlockingSimulation {
+    fun testConnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
         val source = spyk(Source(engine))
@@ -54,7 +54,7 @@ class SimNetworkSwitchVirtualTest {
     }
 
     @Test
-    fun testConnectClosedPort() = runBlockingSimulation {
+    fun testConnectClosedPort() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val sink = SimNetworkSink(engine, capacity = 100.0)
         val switch = SimNetworkSwitchVirtual(engine)
diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt
index 7cc4b801..29c50d3f 100644
--- a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt
@@ -27,17 +27,17 @@ import io.mockk.verify
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.FlowSource
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [SimPdu] class.
  */
 internal class SimPduTest {
     @Test
-    fun testZeroOutlets() = runBlockingSimulation {
+    fun testZeroOutlets() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val pdu = SimPdu(engine)
@@ -47,7 +47,7 @@ internal class SimPduTest {
     }
 
     @Test
-    fun testSingleOutlet() = runBlockingSimulation {
+    fun testSingleOutlet() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val pdu = SimPdu(engine)
@@ -58,7 +58,7 @@ internal class SimPduTest {
     }
 
     @Test
-    fun testDoubleOutlet() = runBlockingSimulation {
+    fun testDoubleOutlet() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val pdu = SimPdu(engine)
@@ -71,7 +71,7 @@ internal class SimPduTest {
     }
 
     @Test
-    fun testDisconnect() = runBlockingSimulation {
+    fun testDisconnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val pdu = SimPdu(engine)
@@ -89,7 +89,7 @@ internal class SimPduTest {
     }
 
     @Test
-    fun testLoss() = runBlockingSimulation {
+    fun testLoss() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         // https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
@@ -100,7 +100,7 @@ internal class SimPduTest {
     }
 
     @Test
-    fun testOutletClose() = runBlockingSimulation {
+    fun testOutletClose() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val pdu = SimPdu(engine)
diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt
index 4f319e65..b83b6ba7 100644
--- a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt
@@ -30,17 +30,17 @@ import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertDoesNotThrow
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.FlowSource
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [SimPowerSource]
  */
 internal class SimPowerSourceTest {
     @Test
-    fun testInitialState() = runBlockingSimulation {
+    fun testInitialState() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
 
@@ -50,7 +50,7 @@ internal class SimPowerSourceTest {
     }
 
     @Test
-    fun testDisconnectIdempotent() = runBlockingSimulation {
+    fun testDisconnectIdempotent() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
 
@@ -59,7 +59,7 @@ internal class SimPowerSourceTest {
     }
 
     @Test
-    fun testConnect() = runBlockingSimulation {
+    fun testConnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val inlet = SimpleInlet()
@@ -74,7 +74,7 @@ internal class SimPowerSourceTest {
     }
 
     @Test
-    fun testDisconnect() = runBlockingSimulation {
+    fun testDisconnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val consumer = spyk(FixedFlowSource(100.0, utilization = 1.0))
@@ -89,7 +89,7 @@ internal class SimPowerSourceTest {
     }
 
     @Test
-    fun testDisconnectAssertion() = runBlockingSimulation {
+    fun testDisconnectAssertion() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val inlet = mockk<SimPowerInlet>(relaxUnitFun = true)
@@ -105,7 +105,7 @@ internal class SimPowerSourceTest {
     }
 
     @Test
-    fun testOutletAlreadyConnected() = runBlockingSimulation {
+    fun testOutletAlreadyConnected() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val inlet = SimpleInlet()
@@ -119,7 +119,7 @@ internal class SimPowerSourceTest {
     }
 
     @Test
-    fun testInletAlreadyConnected() = runBlockingSimulation {
+    fun testInletAlreadyConnected() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val inlet = mockk<SimPowerInlet>(relaxUnitFun = true)
diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt
index e19e72fa..2b2921d7 100644
--- a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt
@@ -27,17 +27,17 @@ import io.mockk.verify
 import org.junit.jupiter.api.Assertions.assertAll
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.FlowSource
 import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.kotlin.runSimulation
 
 /**
  * Test suite for the [SimUps] class.
  */
 internal class SimUpsTest {
     @Test
-    fun testSingleInlet() = runBlockingSimulation {
+    fun testSingleInlet() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         val ups = SimUps(engine)
@@ -48,7 +48,7 @@ internal class SimUpsTest {
     }
 
     @Test
-    fun testDoubleInlet() = runBlockingSimulation {
+    fun testDoubleInlet() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source1 = SimPowerSource(engine, capacity = 100.0)
         val source2 = SimPowerSource(engine, capacity = 100.0)
@@ -65,7 +65,7 @@ internal class SimUpsTest {
     }
 
     @Test
-    fun testLoss() = runBlockingSimulation {
+    fun testLoss() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source = SimPowerSource(engine, capacity = 100.0)
         // https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
@@ -77,7 +77,7 @@ internal class SimUpsTest {
     }
 
     @Test
-    fun testDisconnect() = runBlockingSimulation {
+    fun testDisconnect() = runSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val source1 = SimPowerSource(engine, capacity = 100.0)
         val source2 = SimPowerSource(engine, capacity = 100.0)
diff --git a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
index 74f7c8c1..d340d31c 100644
--- a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
+++ b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt
@@ -33,7 +33,7 @@ import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.power.LinearPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
-import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.kotlin.runSimulation
 import org.opendc.web.proto.runner.Job
 import org.opendc.web.proto.runner.Scenario
 import org.opendc.web.proto.runner.Topology
@@ -222,7 +222,7 @@ public class OpenDCRunner(
         /**
          * Run a single simulation of the scenario.
          */
-        private fun runSimulation(monitor: WebComputeMonitor) = runBlockingSimulation {
+        private fun runSimulation(monitor: WebComputeMonitor) = runSimulation {
             val serviceDomain = "compute.opendc.org"
             val seed = repeat.toLong()
 
diff --git a/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt b/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt
index f8039e1d..e37f489d 100644
--- a/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt
+++ b/opendc-workflow/opendc-workflow-service/src/test/kotlin/org/opendc/workflow/service/WorkflowServiceTest.kt
@@ -44,8 +44,8 @@ import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
-import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.mux.FlowMultiplexerFactory
+import org.opendc.simulator.kotlin.runSimulation
 import org.opendc.trace.Trace
 import org.opendc.workflow.service.scheduler.job.NullJobAdmissionPolicy
 import org.opendc.workflow.service.scheduler.job.SubmissionTimeJobOrderPolicy
@@ -64,7 +64,7 @@ internal class WorkflowServiceTest {
      * A large integration test where we check whether all tasks in some trace are executed correctly.
      */
     @Test
-    fun testTrace() = runBlockingSimulation {
+    fun testTrace() = runSimulation {
         val computeService = "compute.opendc.org"
         val workflowService = "workflow.opendc.org"