refactor(simulator): Migrate to flow-based simulation

This change renames the `opendc-simulator-resources` module into the
`opendc-simulator-flow` module to indicate that the core simulation
model of OpenDC is based around modelling and simulating flows.

Previously, the distinction between resource consumer and provider, and
input and output caused some confusion. By switching to a flow-based
model, this distinction is now clear (as in, the water flows from source
to consumer/sink).

6 files changed, 62 insertions, 61 deletions

diff --git a/opendc-simulator/opendc-simulator-network/build.gradle.kts b/opendc-simulator/opendc-simulator-network/build.gradle.kts
index eb9adcd1..a8f94602 100644
--- a/opendc-simulator/opendc-simulator-network/build.gradle.kts
+++ b/opendc-simulator/opendc-simulator-network/build.gradle.kts
@@ -30,6 +30,6 @@ plugins {
 
 dependencies {
     api(platform(projects.opendcPlatform))
-    api(projects.opendcSimulator.opendcSimulatorResources)
+    api(projects.opendcSimulator.opendcSimulatorFlow)
     implementation(projects.opendcSimulator.opendcSimulatorCore)
 }
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt
index 102e5625..4b66d5cf 100644
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt
+++ b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt
@@ -22,8 +22,8 @@
 
 package org.opendc.simulator.network
 
-import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceProvider
+import org.opendc.simulator.flow.FlowConsumer
+import org.opendc.simulator.flow.FlowSource
 
 /**
  * A network port allows network devices to be connected to network through links.
@@ -78,14 +78,14 @@ public abstract class SimNetworkPort {
     }
 
     /**
-     * Create a [SimResourceConsumer] which generates the outgoing traffic of this port.
+     * Create a [FlowSource] which generates the outgoing traffic of this port.
      */
-    protected abstract fun createConsumer(): SimResourceConsumer
+    protected abstract fun createConsumer(): FlowSource
 
     /**
-     * The [SimResourceProvider] which processes the ingoing traffic of this port.
+     * The [FlowConsumer] which processes the ingoing traffic of this port.
      */
-    protected abstract val provider: SimResourceProvider
+    protected abstract val provider: FlowConsumer
 
     override fun toString(): String = "SimNetworkPort[isConnected=$isConnected]"
 }
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSink.kt b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSink.kt
index 7db0f176..4b0d7bbd 100644
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSink.kt
+++ b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSink.kt
@@ -22,22 +22,22 @@
 
 package org.opendc.simulator.network
 
-import org.opendc.simulator.resources.*
+import org.opendc.simulator.flow.*
 
 /**
  * A network sink which discards all received traffic and does not generate any traffic itself.
  */
 public class SimNetworkSink(
-    interpreter: SimResourceInterpreter,
+    engine: FlowEngine,
     public val capacity: Double
 ) : SimNetworkPort() {
-    override fun createConsumer(): SimResourceConsumer = object : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long = Long.MAX_VALUE
+    override fun createConsumer(): FlowSource = object : FlowSource {
+        override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long = Long.MAX_VALUE
 
         override fun toString(): String = "SimNetworkSink.Consumer"
     }
 
-    override val provider: SimResourceProvider = SimResourceSource(capacity, interpreter)
+    override val provider: FlowConsumer = FlowSink(engine, capacity)
 
     override fun toString(): String = "SimNetworkSink[capacity=$capacity]"
 }
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt
index 2267715e..2b7c1ad7 100644
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt
+++ b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt
@@ -22,12 +22,13 @@
 
 package org.opendc.simulator.network
 
-import org.opendc.simulator.resources.*
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
 
 /**
  * A [SimNetworkSwitch] that can support new networking ports on demand.
  */
-public class SimNetworkSwitchVirtual(interpreter: SimResourceInterpreter) : SimNetworkSwitch {
+public class SimNetworkSwitchVirtual(private val engine: FlowEngine) : SimNetworkSwitch {
     /**
      * The ports of this switch.
      */
@@ -36,9 +37,9 @@ public class SimNetworkSwitchVirtual(interpreter: SimResourceInterpreter) : SimN
     private val _ports = mutableListOf<Port>()
 
     /**
-     * The [SimResourceSwitchMaxMin] to actually perform the switching.
+     * The [MaxMinFlowMultiplexer] to actually perform the switching.
      */
-    private val switch = SimResourceSwitchMaxMin(interpreter)
+    private val mux = MaxMinFlowMultiplexer(engine)
 
     /**
      * Open a new port on the switch.
@@ -58,19 +59,19 @@ public class SimNetworkSwitchVirtual(interpreter: SimResourceInterpreter) : SimN
          */
         private var isClosed: Boolean = false
 
-        override val provider: SimResourceProvider
+        override val provider: FlowConsumer
             get() = _provider
-        private val _provider = switch.newOutput()
+        private val _provider = mux.newInput()
 
-        override fun createConsumer(): SimResourceConsumer {
-            val forwarder = SimResourceForwarder(isCoupled = true)
-            switch.addInput(forwarder)
+        override fun createConsumer(): FlowSource {
+            val forwarder = FlowForwarder(engine, isCoupled = true)
+            mux.addOutput(forwarder)
             return forwarder
         }
 
         override fun close() {
             isClosed = true
-            switch.removeOutput(_provider)
+            mux.removeInput(_provider)
             _ports.remove(this)
         }
     }
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 b8c4b00d..45d0bcf0 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
@@ -31,8 +31,8 @@ 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.resources.*
-import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.source.FixedFlowSource
 
 /**
  * Test suite for the [SimNetworkSink] class.
@@ -40,8 +40,8 @@ import org.opendc.simulator.resources.consumer.SimWorkConsumer
 class SimNetworkSinkTest {
     @Test
     fun testInitialState() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
 
         assertFalse(sink.isConnected)
         assertNull(sink.link)
@@ -50,8 +50,8 @@ class SimNetworkSinkTest {
 
     @Test
     fun testDisconnectIdempotent() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
 
         assertDoesNotThrow { sink.disconnect() }
         assertFalse(sink.isConnected)
@@ -59,8 +59,8 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnectCircular() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
 
         assertThrows<IllegalArgumentException> {
             sink.connect(sink)
@@ -69,8 +69,8 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnectAlreadyConnectedTarget() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
         val source = mockk<SimNetworkPort>(relaxUnitFun = true)
         every { source.isConnected } returns true
 
@@ -81,9 +81,9 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnectAlreadyConnected() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
-        val source1 = Source(interpreter)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val source1 = Source(engine)
 
         val source2 = mockk<SimNetworkPort>(relaxUnitFun = true)
 
@@ -97,9 +97,9 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnect() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
-        val source = spyk(Source(interpreter))
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val source = spyk(Source(engine))
         val consumer = source.consumer
 
         sink.connect(source)
@@ -108,14 +108,14 @@ class SimNetworkSinkTest {
         assertTrue(source.isConnected)
 
         verify { source.createConsumer() }
-        verify { consumer.onEvent(any(), SimResourceEvent.Start) }
+        verify { consumer.onEvent(any(), any(), FlowEvent.Start) }
     }
 
     @Test
     fun testDisconnect() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
-        val source = spyk(Source(interpreter))
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val source = spyk(Source(engine))
         val consumer = source.consumer
 
         sink.connect(source)
@@ -124,14 +124,14 @@ class SimNetworkSinkTest {
         assertFalse(sink.isConnected)
         assertFalse(source.isConnected)
 
-        verify { consumer.onEvent(any(), SimResourceEvent.Exit) }
+        verify { consumer.onEvent(any(), any(), FlowEvent.Exit) }
     }
 
-    private class Source(interpreter: SimResourceInterpreter) : SimNetworkPort() {
-        val consumer = spyk(SimWorkConsumer(Double.POSITIVE_INFINITY, utilization = 0.8))
+    private class Source(engine: FlowEngine) : SimNetworkPort() {
+        val consumer = spyk(FixedFlowSource(Double.POSITIVE_INFINITY, utilization = 0.8))
 
-        public override fun createConsumer(): SimResourceConsumer = consumer
+        public override fun createConsumer(): FlowSource = consumer
 
-        override val provider: SimResourceProvider = SimResourceSource(0.0, interpreter)
+        override val provider: FlowConsumer = FlowSink(engine, 0.0)
     }
 }
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 3a749bfe..4aa2fa92 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
@@ -28,8 +28,8 @@ 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.resources.*
-import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.source.FixedFlowSource
 
 /**
  * Test suite for the [SimNetworkSwitchVirtual] class.
@@ -37,10 +37,10 @@ import org.opendc.simulator.resources.consumer.SimWorkConsumer
 class SimNetworkSwitchVirtualTest {
     @Test
     fun testConnect() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
-        val source = spyk(Source(interpreter))
-        val switch = SimNetworkSwitchVirtual(interpreter)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val source = spyk(Source(engine))
+        val switch = SimNetworkSwitchVirtual(engine)
         val consumer = source.consumer
 
         switch.newPort().connect(sink)
@@ -50,14 +50,14 @@ class SimNetworkSwitchVirtualTest {
         assertTrue(source.isConnected)
 
         verify { source.createConsumer() }
-        verify { consumer.onEvent(any(), SimResourceEvent.Start) }
+        verify { consumer.onEvent(any(), any(), FlowEvent.Start) }
     }
 
     @Test
     fun testConnectClosedPort() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val sink = SimNetworkSink(interpreter, capacity = 100.0)
-        val switch = SimNetworkSwitchVirtual(interpreter)
+        val engine = FlowEngine(coroutineContext, clock)
+        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val switch = SimNetworkSwitchVirtual(engine)
 
         val port = switch.newPort()
         port.close()
@@ -67,11 +67,11 @@ class SimNetworkSwitchVirtualTest {
         }
     }
 
-    private class Source(interpreter: SimResourceInterpreter) : SimNetworkPort() {
-        val consumer = spyk(SimWorkConsumer(Double.POSITIVE_INFINITY, utilization = 0.8))
+    private class Source(engine: FlowEngine) : SimNetworkPort() {
+        val consumer = spyk(FixedFlowSource(Double.POSITIVE_INFINITY, utilization = 0.8))
 
-        public override fun createConsumer(): SimResourceConsumer = consumer
+        public override fun createConsumer(): FlowSource = consumer
 
-        override val provider: SimResourceProvider = SimResourceSource(0.0, interpreter)
+        override val provider: FlowConsumer = FlowSink(engine, 0.0)
     }
 }