simulator: Add uniform interface for resource metrics

This change adds a new interface to the resources library for accessing
metrics of resources such as work, demand and overcommitted work. With
this change, we do not need an implementation specific listener
interface in SimResourceSwitchMaxMin anymore.

Another benefit of this approach is that updates will be scheduled more
efficiently and progress will only be reported once the system has
reached a steady-state for that timestamp.

4 files changed, 72 insertions, 69 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
index 994ae888..51024e80 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMinTest.kt
@@ -59,7 +59,7 @@ internal class SimResourceAggregatorMaxMinTest {
         source.startConsumer(adapter)
 
         try {
-            aggregator.output.consume(consumer)
+            aggregator.consume(consumer)
             yield()
 
             assertAll(
@@ -67,7 +67,7 @@ internal class SimResourceAggregatorMaxMinTest {
                 { assertEquals(listOf(0.0, 0.5, 0.0), usage) }
             )
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
@@ -87,14 +87,14 @@ internal class SimResourceAggregatorMaxMinTest {
         val adapter = SimSpeedConsumerAdapter(consumer, usage::add)
 
         try {
-            aggregator.output.consume(adapter)
+            aggregator.consume(adapter)
             yield()
             assertAll(
                 { assertEquals(1000, clock.millis()) },
                 { assertEquals(listOf(0.0, 2.0, 0.0), usage) }
             )
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
@@ -115,13 +115,13 @@ internal class SimResourceAggregatorMaxMinTest {
             .andThen(SimResourceCommand.Exit)
 
         try {
-            aggregator.output.consume(consumer)
+            aggregator.consume(consumer)
             yield()
             assertEquals(1000, clock.millis())
 
             verify(exactly = 2) { consumer.onNext(any()) }
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
@@ -142,11 +142,11 @@ internal class SimResourceAggregatorMaxMinTest {
             .andThenThrows(IllegalStateException("Test Exception"))
 
         try {
-            assertThrows<IllegalStateException> { aggregator.output.consume(consumer) }
+            assertThrows<IllegalStateException> { aggregator.consume(consumer) }
             yield()
             assertEquals(SimResourceState.Pending, sources[0].state)
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
@@ -164,14 +164,14 @@ internal class SimResourceAggregatorMaxMinTest {
         val consumer = SimWorkConsumer(4.0, 1.0)
         try {
             coroutineScope {
-                launch { aggregator.output.consume(consumer) }
+                launch { aggregator.consume(consumer) }
                 delay(1000)
                 sources[0].capacity = 0.5
             }
             yield()
             assertEquals(2334, clock.millis())
         } finally {
-            aggregator.output.close()
+            aggregator.close()
         }
     }
 
@@ -189,14 +189,40 @@ internal class SimResourceAggregatorMaxMinTest {
         val consumer = SimWorkConsumer(1.0, 0.5)
         try {
             coroutineScope {
-                launch { aggregator.output.consume(consumer) }
+                launch { aggregator.consume(consumer) }
                 delay(500)
                 sources[0].capacity = 0.5
             }
             yield()
             assertEquals(1000, clock.millis())
         } finally {
-            aggregator.output.close()
+            aggregator.close()
+        }
+    }
+
+    @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 = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) }
+            .returns(SimResourceCommand.Consume(4.0, 4.0, 1000))
+            .andThen(SimResourceCommand.Exit)
+
+        try {
+            aggregator.consume(consumer)
+            yield()
+            assertEquals(1000, clock.millis())
+            assertEquals(2.0, aggregator.counters.actual) { "Actual work mismatch" }
+        } finally {
+            aggregator.close()
         }
     }
 }
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 517dcb36..ad8d82e3 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
@@ -67,7 +67,7 @@ internal class SimResourceSwitchExclusiveTest {
         source.startConsumer(adapter)
         switch.addInput(forwarder)
 
-        val provider = switch.addOutput(3200.0)
+        val provider = switch.newOutput()
 
         try {
             provider.consume(workload)
@@ -98,7 +98,7 @@ internal class SimResourceSwitchExclusiveTest {
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(3200.0)
+        val provider = switch.newOutput()
 
         try {
             provider.consume(workload)
@@ -142,7 +142,7 @@ internal class SimResourceSwitchExclusiveTest {
 
         switch.addInput(source)
 
-        val provider = switch.addOutput(3200.0)
+        val provider = switch.newOutput()
 
         try {
             provider.consume(workload)
@@ -170,7 +170,7 @@ internal class SimResourceSwitchExclusiveTest {
 
         switch.addInput(source)
 
-        switch.addOutput(3200.0)
-        assertThrows<IllegalStateException> { switch.addOutput(3200.0) }
+        switch.newOutput()
+        assertThrows<IllegalStateException> { switch.newOutput() }
     }
 }
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 0b023878..e4292ec0 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
@@ -47,7 +47,7 @@ internal class SimResourceSwitchMaxMinTest {
         val sources = List(2) { SimResourceSource(2000.0, scheduler) }
         sources.forEach { switch.addInput(it) }
 
-        val provider = switch.addOutput(1000.0)
+        val provider = switch.newOutput()
 
         val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
         every { consumer.onNext(any()) } returns SimResourceCommand.Consume(1.0, 1.0) andThen SimResourceCommand.Exit
@@ -67,26 +67,6 @@ internal class SimResourceSwitchMaxMinTest {
     fun testOvercommittedSingle() = runBlockingSimulation {
         val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
-        val listener = object : SimResourceSwitchMaxMin.Listener {
-            var totalRequestedWork = 0L
-            var totalGrantedWork = 0L
-            var totalOvercommittedWork = 0L
-
-            override fun onSliceFinish(
-                switch: SimResourceSwitchMaxMin,
-                requestedWork: Long,
-                grantedWork: Long,
-                overcommittedWork: Long,
-                interferedWork: Long,
-                cpuUsage: Double,
-                cpuDemand: Double
-            ) {
-                totalRequestedWork += requestedWork
-                totalGrantedWork += grantedWork
-                totalOvercommittedWork += overcommittedWork
-            }
-        }
-
         val duration = 5 * 60L
         val workload =
             SimTraceConsumer(
@@ -98,8 +78,8 @@ internal class SimResourceSwitchMaxMinTest {
                 ),
             )
 
-        val switch = SimResourceSwitchMaxMin(scheduler, null, listener)
-        val provider = switch.addOutput(3200.0)
+        val switch = SimResourceSwitchMaxMin(scheduler)
+        val provider = switch.newOutput()
 
         try {
             switch.addInput(SimResourceSource(3200.0, scheduler))
@@ -110,9 +90,9 @@ internal class SimResourceSwitchMaxMinTest {
         }
 
         assertAll(
-            { assertEquals(1113300, listener.totalRequestedWork, "Requested Burst does not match") },
-            { assertEquals(1023300, listener.totalGrantedWork, "Granted Burst does not match") },
-            { assertEquals(90000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(1113300.0, switch.counters.demand, "Requested work does not match") },
+            { assertEquals(1023300.0, switch.counters.actual, "Actual work does not match") },
+            { assertEquals(90000.0, switch.counters.overcommit, "Overcommitted work does not match") },
             { assertEquals(1200000, clock.millis()) }
         )
     }
@@ -124,26 +104,6 @@ internal class SimResourceSwitchMaxMinTest {
     fun testOvercommittedDual() = runBlockingSimulation {
         val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
 
-        val listener = object : SimResourceSwitchMaxMin.Listener {
-            var totalRequestedWork = 0L
-            var totalGrantedWork = 0L
-            var totalOvercommittedWork = 0L
-
-            override fun onSliceFinish(
-                switch: SimResourceSwitchMaxMin,
-                requestedWork: Long,
-                grantedWork: Long,
-                overcommittedWork: Long,
-                interferedWork: Long,
-                cpuUsage: Double,
-                cpuDemand: Double
-            ) {
-                totalRequestedWork += requestedWork
-                totalGrantedWork += grantedWork
-                totalOvercommittedWork += overcommittedWork
-            }
-        }
-
         val duration = 5 * 60L
         val workloadA =
             SimTraceConsumer(
@@ -164,9 +124,9 @@ internal class SimResourceSwitchMaxMinTest {
                 )
             )
 
-        val switch = SimResourceSwitchMaxMin(scheduler, null, listener)
-        val providerA = switch.addOutput(3200.0)
-        val providerB = switch.addOutput(3200.0)
+        val switch = SimResourceSwitchMaxMin(scheduler)
+        val providerA = switch.newOutput()
+        val providerB = switch.newOutput()
 
         try {
             switch.addInput(SimResourceSource(3200.0, scheduler))
@@ -181,9 +141,9 @@ internal class SimResourceSwitchMaxMinTest {
             switch.close()
         }
         assertAll(
-            { assertEquals(2073600, listener.totalRequestedWork, "Requested Burst does not match") },
-            { assertEquals(1053600, listener.totalGrantedWork, "Granted Burst does not match") },
-            { assertEquals(1020000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(2073600.0, switch.counters.demand, "Requested work does not match") },
+            { assertEquals(1053600.0, switch.counters.actual, "Granted work does not match") },
+            { assertEquals(1020000.0, switch.counters.overcommit, "Overcommitted work does not match") },
             { assertEquals(1200000, clock.millis()) }
         )
     }
diff --git a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt
index 04886399..810052b8 100644
--- a/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceTransformerTest.kt
@@ -206,4 +206,21 @@ internal class SimResourceTransformerTest {
         assertEquals(0, clock.millis())
         verify(exactly = 1) { consumer.onNext(any()) }
     }
+
+    @Test
+    fun testCounters() = runBlockingSimulation {
+        val forwarder = SimResourceForwarder()
+        val scheduler = SimResourceInterpreterImpl(coroutineContext, clock)
+        val source = SimResourceSource(1.0, scheduler)
+
+        val consumer = SimWorkConsumer(2.0, 1.0)
+        source.startConsumer(forwarder)
+
+        forwarder.consume(consumer)
+
+        assertEquals(source.counters.actual, forwarder.counters.actual) { "Actual work" }
+        assertEquals(source.counters.demand, forwarder.counters.demand) { "Work demand" }
+        assertEquals(source.counters.overcommit, forwarder.counters.overcommit) { "Overcommitted work" }
+        assertEquals(2000, clock.millis())
+    }
 }