refactor(simulator): Merge distributor and aggregator into switch

This change removes the distributor and aggregator interfaces in favour
of a single switch interface. Since the switch interface is as powerful
as both the distributor and aggregator, we don't need the latter two.

3 files changed, 10 insertions, 214 deletions

diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
deleted file mode 100644
index f4ea5fe8..00000000
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
+++ /dev/null
@@ -1,190 +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.resources
-
-import io.mockk.spyk
-import io.mockk.verify
-import kotlinx.coroutines.*
-import org.junit.jupiter.api.Assertions.*
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.assertAll
-import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.simulator.resources.consumer.SimSpeedConsumerAdapter
-import org.opendc.simulator.resources.consumer.SimWorkConsumer
-import org.opendc.simulator.resources.impl.SimResourceInterpreterImpl
-
-/**
- * Test suite for the [SimResourceAggregatorMaxMin] class.
- */
-internal class SimResourceAggregatorMaxMinTest {
-    @Test
-    fun testSingleCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-
-        val aggregator = SimResourceAggregatorMaxMin(scheduler)
-        val forwarder = SimResourceForwarder()
-        val sources = listOf(
-            forwarder,
-            SimResourceSource(1.0, scheduler)
-        )
-        sources.forEach(aggregator::addInput)
-
-        val consumer = SimWorkConsumer(1.0, 0.5)
-        val usage = mutableListOf<Double>()
-        val source = SimResourceSource(1.0, scheduler)
-        val adapter = SimSpeedConsumerAdapter(forwarder, usage::add)
-        source.startConsumer(adapter)
-
-        aggregator.consume(consumer)
-        yield()
-
-        assertAll(
-            { assertEquals(1000, clock.millis()) },
-            { assertEquals(listOf(0.0, 0.5, 0.0), usage) }
-        )
-    }
-
-    @Test
-    fun testDoubleCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-
-        val aggregator = SimResourceAggregatorMaxMin(scheduler)
-        val sources = listOf(
-            SimResourceSource(1.0, scheduler),
-            SimResourceSource(1.0, scheduler)
-        )
-        sources.forEach(aggregator::addInput)
-
-        val consumer = SimWorkConsumer(2.0, 1.0)
-        val usage = mutableListOf<Double>()
-        val adapter = SimSpeedConsumerAdapter(consumer, usage::add)
-
-        aggregator.consume(adapter)
-        yield()
-        assertAll(
-            { assertEquals(1000, clock.millis()) },
-            { assertEquals(listOf(0.0, 2.0, 0.0), usage) }
-        )
-    }
-
-    @Test
-    fun testOvercommit() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-
-        val aggregator = SimResourceAggregatorMaxMin(scheduler)
-        val sources = listOf(
-            SimResourceSource(1.0, scheduler),
-            SimResourceSource(1.0, scheduler)
-        )
-        sources.forEach(aggregator::addInput)
-
-        val consumer = spyk(object : SimResourceConsumer {
-            override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
-                return if (now == 0L) {
-                    ctx.push(4.0)
-                    1000
-                } else {
-                    ctx.close()
-                    Long.MAX_VALUE
-                }
-            }
-        })
-
-        aggregator.consume(consumer)
-        yield()
-        assertEquals(1000, clock.millis())
-
-        verify(exactly = 2) { consumer.onNext(any(), any(), any()) }
-    }
-
-    @Test
-    fun testAdjustCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-
-        val aggregator = SimResourceAggregatorMaxMin(scheduler)
-        val sources = listOf(
-            SimResourceSource(1.0, scheduler),
-            SimResourceSource(1.0, scheduler)
-        )
-        sources.forEach(aggregator::addInput)
-
-        val consumer = SimWorkConsumer(4.0, 1.0)
-        coroutineScope {
-            launch { aggregator.consume(consumer) }
-            delay(1000)
-            sources[0].capacity = 0.5
-        }
-        yield()
-        assertEquals(2333, clock.millis())
-    }
-
-    @Test
-    fun testFailOverCapacity() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-
-        val aggregator = SimResourceAggregatorMaxMin(scheduler)
-        val sources = listOf(
-            SimResourceSource(1.0, scheduler),
-            SimResourceSource(1.0, scheduler)
-        )
-        sources.forEach(aggregator::addInput)
-
-        val consumer = SimWorkConsumer(1.0, 0.5)
-        coroutineScope {
-            launch { aggregator.consume(consumer) }
-            delay(500)
-            sources[0].capacity = 0.5
-        }
-        yield()
-        assertEquals(1167, clock.millis())
-    }
-
-    @Test
-    fun testCounters() = runBlockingSimulation {
-        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
-
-        val aggregator = SimResourceAggregatorMaxMin(scheduler)
-        val sources = listOf(
-            SimResourceSource(1.0, scheduler),
-            SimResourceSource(1.0, scheduler)
-        )
-        sources.forEach(aggregator::addInput)
-
-        val consumer = object : SimResourceConsumer {
-            override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
-                return if (now == 0L) {
-                    ctx.push(4.0)
-                    1000
-                } else {
-                    ctx.close()
-                    Long.MAX_VALUE
-                }
-            }
-        }
-
-        aggregator.consume(consumer)
-        yield()
-        assertEquals(1000, clock.millis())
-        assertEquals(2.0, aggregator.counters.actual) { "Actual work mismatch" }
-    }
-}
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
index 9f86dc0d..49f2da5f 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
@@ -65,13 +65,8 @@ internal class SimResourceSwitchExclusiveTest {
         switch.addInput(forwarder)
 
         val provider = switch.newOutput()
-
-        try {
-            provider.consume(workload)
-            yield()
-        } finally {
-            provider.close()
-        }
+        provider.consume(workload)
+        yield()
 
         assertAll(
             { assertEquals(listOf(0.0, 28.0, 3200.0, 0.0, 183.0, 0.0), speed) { "Correct speed" } },
@@ -95,13 +90,9 @@ internal class SimResourceSwitchExclusiveTest {
         switch.addInput(source)
 
         val provider = switch.newOutput()
+        provider.consume(workload)
+        yield()
 
-        try {
-            provider.consume(workload)
-            yield()
-        } finally {
-            provider.close()
-        }
         assertEquals(duration, clock.millis()) { "Took enough time" }
     }
 
@@ -141,14 +132,9 @@ internal class SimResourceSwitchExclusiveTest {
         switch.addInput(source)
 
         val provider = switch.newOutput()
-
-        try {
-            provider.consume(workload)
-            yield()
-            provider.consume(workload)
-        } finally {
-            provider.close()
-        }
+        provider.consume(workload)
+        yield()
+        provider.consume(workload)
         assertEquals(duration * 2, clock.millis()) { "Took enough time" }
     }
 
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
index ba0d66ff..03f90e21 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
@@ -51,7 +51,7 @@ internal class SimResourceSwitchMaxMinTest {
             provider.consume(consumer)
             yield()
         } finally {
-            switch.close()
+            switch.clear()
         }
     }
 
@@ -81,7 +81,7 @@ internal class SimResourceSwitchMaxMinTest {
             provider.consume(workload)
             yield()
         } finally {
-            switch.close()
+            switch.clear()
         }
 
         assertAll(
@@ -133,7 +133,7 @@ internal class SimResourceSwitchMaxMinTest {
 
             yield()
         } finally {
-            switch.close()
+            switch.clear()
         }
         assertAll(
             { assertEquals(2073600.0, switch.counters.demand, "Requested work does not match") },