simulator: Add support for signaling dynamic capacity changes

This change adds support for dynamically changing the capacity of
resources and propagating this change to consumers.

4 files changed, 105 insertions, 18 deletions

diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
index 3dffc7bf..de864c1c 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
@@ -25,11 +25,9 @@ package org.opendc.simulator.resources
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.verify
-import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.toList
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runBlockingTest
-import kotlinx.coroutines.yield
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertAll
@@ -155,4 +153,56 @@ internal class SimResourceAggregatorMaxMinTest {
             aggregator.output.close()
         }
     }
+
+    @Test
+    fun testAdjustCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = SimWorkConsumer(4.0, 1.0)
+        try {
+            coroutineScope {
+                launch { aggregator.output.consume(consumer) }
+                delay(1000)
+                sources[0].capacity = 0.5
+            }
+            yield()
+            assertEquals(2334, currentTime)
+        } finally {
+            aggregator.output.close()
+        }
+    }
+
+    @Test
+    fun testFailOverCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val aggregator = SimResourceAggregatorMaxMin(clock)
+        val sources = listOf(
+            SimResourceSource(1.0, clock, scheduler),
+            SimResourceSource(1.0, clock, scheduler)
+        )
+        sources.forEach(aggregator::addInput)
+
+        val consumer = SimWorkConsumer(1.0, 0.5)
+        try {
+            coroutineScope {
+                launch { aggregator.output.consume(consumer) }
+                delay(500)
+                sources[0].capacity = 0.5
+            }
+            yield()
+            assertEquals(1000, currentTime)
+        } finally {
+            aggregator.output.close()
+        }
+    }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
index c6988ed9..030a0f6b 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
@@ -40,9 +40,7 @@ class SimResourceContextTest {
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-        val context = object : SimAbstractResourceContext(clock, consumer) {
-            override val capacity: Double = 4200.0
-
+        val context = object : SimAbstractResourceContext(4200.0, clock, consumer) {
             override fun onIdle(deadline: Long) {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
             override fun onFinish(cause: Throwable?) {}
@@ -58,9 +56,7 @@ class SimResourceContextTest {
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(10.0, 1.0) andThen SimResourceCommand.Exit
 
-        val context = spyk(object : SimAbstractResourceContext(clock, consumer) {
-            override val capacity: Double = 4200.0
-
+        val context = spyk(object : SimAbstractResourceContext(4200.0, clock, consumer) {
             override fun onIdle(deadline: Long) {}
             override fun onFinish(cause: Throwable?) {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
@@ -80,9 +76,7 @@ class SimResourceContextTest {
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10) andThen SimResourceCommand.Exit
 
-        val context = spyk(object : SimAbstractResourceContext(clock, consumer) {
-            override val capacity: Double = 4200.0
-
+        val context = spyk(object : SimAbstractResourceContext(4200.0, clock, consumer) {
             override fun onIdle(deadline: Long) {}
             override fun onFinish(cause: Throwable?) {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
@@ -107,9 +101,7 @@ class SimResourceContextTest {
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Idle(10) andThen SimResourceCommand.Exit
 
-        val context = object : SimAbstractResourceContext(clock, consumer) {
-            override val capacity: Double = 4200.0
-
+        val context = object : SimAbstractResourceContext(4200.0, clock, consumer) {
             override fun onIdle(deadline: Long) {}
             override fun onFinish(cause: Throwable?) {}
             override fun onConsume(work: Double, limit: Double, deadline: Long) {}
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
index f68450ff..143dbca9 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
@@ -24,14 +24,14 @@ package org.opendc.simulator.resources
 
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.spyk
 import io.mockk.verify
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.*
 import kotlinx.coroutines.test.runBlockingTest
-import kotlinx.coroutines.yield
 import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
 
@@ -163,4 +163,24 @@ internal class SimResourceForwarderTest {
         verify(exactly = 1) { consumer.onStart(any()) }
         verify(exactly = 1) { consumer.onFinish(any(), null) }
     }
+
+    @Test
+    fun testAdjustCapacity() = runBlockingTest {
+        val forwarder = SimResourceForwarder()
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(1.0, clock, scheduler)
+
+        val consumer = spyk(SimWorkConsumer(2.0, 1.0))
+        source.startConsumer(forwarder)
+
+        coroutineScope {
+            launch { forwarder.consume(consumer) }
+            delay(1000)
+            source.capacity = 0.5
+        }
+
+        assertEquals(3000, currentTime)
+        verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
+    }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
index 1279c679..58e19421 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
@@ -24,11 +24,14 @@ package org.opendc.simulator.resources
 
 import io.mockk.every
 import io.mockk.mockk
+import io.mockk.spyk
+import io.mockk.verify
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
 
@@ -64,6 +67,28 @@ class SimResourceSourceTest {
     }
 
     @Test
+    fun testAdjustCapacity() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(1.0, clock, scheduler)
+
+        val consumer = spyk(SimWorkConsumer(2.0, 1.0))
+
+        try {
+            coroutineScope {
+                launch { provider.consume(consumer) }
+                delay(1000)
+                provider.capacity = 0.5
+            }
+            assertEquals(3000, currentTime)
+            verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
+        } finally {
+            scheduler.close()
+            provider.close()
+        }
+    }
+
+    @Test
     fun testSpeedLimit() = runBlockingTest {
         val clock = DelayControllerClockAdapter(this)
         val scheduler = TimerScheduler<Any>(coroutineContext, clock)