simulator: Re-design consumer interface to support capacity negotiation

This change re-designs the SimResourceConsumer interface to support in
the future capacity negotiation. This basically means that the consumer
will be informed directly when not enough capacity is available, instead
of after the deadline specified by the consumer.

7 files changed, 10 insertions, 57 deletions

diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
index a99b082a..281d43ae 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
@@ -32,7 +32,6 @@ import org.opendc.simulator.compute.model.SimProcessingUnit
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.*
 import java.time.Clock
-import kotlin.coroutines.CoroutineContext
 
 /**
  * Abstract implementation of the [SimHypervisor] interface.
@@ -121,7 +120,7 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
                     override val clock: Clock
                         get() = this@SimAbstractHypervisor.context.clock
 
-                    override val meta: Map<String, Any> = meta + mapOf("coroutine-context" to context.meta["coroutine-context"] as CoroutineContext)
+                    override val meta: Map<String, Any> = meta
 
                     override fun interrupt(resource: SimResource) {
                         requireNotNull(this@VirtualMachine.cpus[resource]).interrupt()
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
index 39ae34fe..44906c2b 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
@@ -33,6 +33,7 @@ import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.SimResource
 import org.opendc.simulator.resources.SimResourceProvider
 import org.opendc.simulator.resources.SimResourceSource
+import org.opendc.simulator.resources.consume
 import java.time.Clock
 import kotlin.coroutines.CoroutineContext
 
@@ -91,7 +92,7 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
     override suspend fun run(workload: SimWorkload, meta: Map<String, Any>): Unit = withContext(context) {
         val resources = resources
         require(!isTerminated) { "Machine is terminated" }
-        val ctx = Context(resources, meta + mapOf("coroutine-context" to context))
+        val ctx = Context(resources, meta)
         val totalCapacity = model.cpus.sumByDouble { it.frequency }
 
         _speed = MutableList(model.cpus.size) { 0.0 }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
index bb97192d..c629fbd9 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
@@ -25,7 +25,6 @@ package org.opendc.simulator.compute
 import org.opendc.simulator.compute.model.SimProcessingUnit
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.*
-import kotlin.coroutines.CoroutineContext
 
 /**
  * A [SimHypervisor] that distributes the computing requirements of multiple [SimWorkload] on a single
@@ -40,7 +39,6 @@ public class SimFairShareHypervisor(private val listener: SimHypervisor.Listener
     override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch<SimProcessingUnit> {
         return SimResourceSwitchMaxMin(
             ctx.clock,
-            ctx.meta["coroutine-context"] as CoroutineContext,
             object : SimResourceSwitchMaxMin.Listener<SimProcessingUnit> {
                 override fun onSliceFinish(
                     switch: SimResourceSwitchMaxMin<SimProcessingUnit>,
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
index 2001a230..5de69884 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
@@ -24,7 +24,6 @@ package org.opendc.simulator.compute
 
 import org.opendc.simulator.compute.model.SimProcessingUnit
 import org.opendc.simulator.resources.*
-import kotlin.coroutines.CoroutineContext
 
 /**
  * A [SimHypervisor] that allocates its sub-resources exclusively for the virtual machine that it hosts.
@@ -35,6 +34,6 @@ public class SimSpaceSharedHypervisor : SimAbstractHypervisor() {
     }
 
     override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch<SimProcessingUnit> {
-        return SimResourceSwitchExclusive(ctx.meta["coroutine-context"] as CoroutineContext)
+        return SimResourceSwitchExclusive()
     }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
index 9b47821e..f1079ee6 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
@@ -24,9 +24,8 @@ package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.compute.model.SimProcessingUnit
-import org.opendc.simulator.resources.SimResourceCommand
 import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
  * A [SimWorkload] that models applications as a static number of floating point operations ([flops]) executed on
@@ -47,29 +46,7 @@ public class SimFlopsWorkload(
     override fun onStart(ctx: SimMachineContext) {}
 
     override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit> {
-        return CpuConsumer(ctx)
-    }
-
-    private inner class CpuConsumer(private val machine: SimMachineContext) : SimResourceConsumer<SimProcessingUnit> {
-        override fun onStart(ctx: SimResourceContext<SimProcessingUnit>): SimResourceCommand {
-            val limit = ctx.resource.frequency * utilization
-            val work = flops.toDouble() / machine.cpus.size
-
-            return if (work > 0.0) {
-                SimResourceCommand.Consume(work, limit)
-            } else {
-                SimResourceCommand.Exit
-            }
-        }
-
-        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, remainingWork: Double): SimResourceCommand {
-            return if (remainingWork > 0.0) {
-                val limit = ctx.resource.frequency * utilization
-                return SimResourceCommand.Consume(remainingWork, limit)
-            } else {
-                SimResourceCommand.Exit
-            }
-        }
+        return SimWorkConsumer(flops.toDouble() / ctx.cpus.size, utilization)
     }
 
     override fun toString(): String = "SimFlopsWorkload(FLOPs=$flops,utilization=$utilization)"
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
index 313b6ed5..d7aa8f80 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
@@ -24,9 +24,8 @@ package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.compute.model.SimProcessingUnit
-import org.opendc.simulator.resources.SimResourceCommand
 import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
  * A [SimWorkload] that models application execution as a single duration.
@@ -46,24 +45,8 @@ public class SimRuntimeWorkload(
     override fun onStart(ctx: SimMachineContext) {}
 
     override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit> {
-        return CpuConsumer()
-    }
-
-    private inner class CpuConsumer : SimResourceConsumer<SimProcessingUnit> {
-        override fun onStart(ctx: SimResourceContext<SimProcessingUnit>): SimResourceCommand {
-            val limit = ctx.resource.frequency * utilization
-            val work = (limit / 1000) * duration
-            return SimResourceCommand.Consume(work, limit)
-        }
-
-        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, remainingWork: Double): SimResourceCommand {
-            return if (remainingWork > 0.0) {
-                val limit = ctx.resource.frequency * utilization
-                SimResourceCommand.Consume(remainingWork, limit)
-            } else {
-                SimResourceCommand.Exit
-            }
-        }
+        val limit = cpu.frequency * utilization
+        return SimWorkConsumer((limit / 1000) * duration, utilization)
     }
 
     override fun toString(): String = "SimRuntimeWorkload(duration=$duration,utilization=$utilization)"
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
index 31f58a0f..cc4f3136 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
@@ -50,11 +50,7 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>) : SimWorkloa
     }
 
     private inner class CpuConsumer : SimResourceConsumer<SimProcessingUnit> {
-        override fun onStart(ctx: SimResourceContext<SimProcessingUnit>): SimResourceCommand {
-            return onNext(ctx, 0.0)
-        }
-
-        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, remainingWork: Double): SimResourceCommand {
+        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, capacity: Double, remainingWork: Double): SimResourceCommand {
             val now = ctx.clock.millis()
             val fragment = fragment ?: return SimResourceCommand.Exit
             val work = (fragment.duration / 1000) * fragment.usage