Add configurable resource distribution at host level (#355)

4 files changed, 76 insertions, 8 deletions

diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
index 791ab692..ced38480 100644
--- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
+++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/provisioner/HostsProvisioningStep.kt
@@ -33,8 +33,9 @@ import org.opendc.simulator.compute.power.SimPowerSource
 import org.opendc.simulator.compute.power.batteries.BatteryAggregator
 import org.opendc.simulator.compute.power.batteries.SimBattery
 import org.opendc.simulator.engine.engine.FlowEngine
-import org.opendc.simulator.engine.graph.FlowDistributor
 import org.opendc.simulator.engine.graph.FlowEdge
+import org.opendc.simulator.engine.graph.distributionPolicies.FlowDistributorFactory
+import org.opendc.simulator.engine.graph.distributionPolicies.FlowDistributorFactory.DistributionPolicy
 
 /**
  * A [ProvisioningStep] that provisions a list of hosts for a [ComputeService].
@@ -66,7 +67,11 @@ public class HostsProvisioningStep internal constructor(
             simPowerSources.add(simPowerSource)
             service.addPowerSource(simPowerSource)
 
-            val hostDistributor = FlowDistributor(engine)
+            val hostDistributor =
+                FlowDistributorFactory.getFlowDistributor(
+                    engine,
+                    DistributionPolicy.MAX_MIN_FAIRNESS,
+                )
 
             val carbonFragments = getCarbonFragments(cluster.powerSource.carbonTracePath)
 
@@ -80,7 +85,11 @@ public class HostsProvisioningStep internal constructor(
 
             if (cluster.battery != null) {
                 // Create Battery Distributor
-                val batteryDistributor = FlowDistributor(engine)
+                val batteryDistributor =
+                    FlowDistributorFactory.getFlowDistributor(
+                        engine,
+                        DistributionPolicy.MAX_MIN_FAIRNESS,
+                    )
                 FlowEdge(batteryDistributor, simPowerSource)
 
                 // Create Battery
diff --git a/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/TopologyFactories.kt b/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/TopologyFactories.kt
index 09a8fe64..c1617304 100644
--- a/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/TopologyFactories.kt
+++ b/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/TopologyFactories.kt
@@ -31,13 +31,12 @@ import org.opendc.compute.topology.specs.HostJSONSpec
 import org.opendc.compute.topology.specs.HostSpec
 import org.opendc.compute.topology.specs.PowerSourceSpec
 import org.opendc.compute.topology.specs.TopologySpec
+import org.opendc.compute.topology.specs.toDistributionPolicy
 import org.opendc.simulator.compute.models.CpuModel
 import org.opendc.simulator.compute.models.GpuModel
 import org.opendc.simulator.compute.models.MachineModel
 import org.opendc.simulator.compute.models.MemoryUnit
 import org.opendc.simulator.compute.power.getPowerModel
-import org.opendc.simulator.engine.graph.distributionPolicies.DistributionPolicyFactory
-import org.opendc.simulator.engine.graph.distributionPolicies.DistributionPolicyFactory.DistributionPolicyType
 import java.io.File
 import java.io.InputStream
 
@@ -189,9 +188,8 @@ private fun HostJSONSpec.toHostSpec(clusterName: String): HostSpec {
             units,
             unknownMemoryUnit,
             gpuUnits,
-            // TODO: Pass through
-            DistributionPolicyFactory.getDistributionStrategy(DistributionPolicyType.MaxMinFairness),
-            DistributionPolicyFactory.getDistributionStrategy(DistributionPolicyType.MaxMinFairness),
+            cpuDistributionPolicy.toDistributionPolicy(),
+            gpuDistributionPolicy.toDistributionPolicy(),
         )
 
     val cpuPowerModel =
@@ -227,6 +225,8 @@ private fun HostJSONSpec.toHostSpec(clusterName: String): HostSpec {
             machineModel,
             cpuPowerModel,
             gpuPowerModel,
+            cpuDistributionPolicy = cpuDistributionPolicy.toDistributionPolicy(),
+            gpuDistributionPolicy = gpuDistributionPolicy.toDistributionPolicy(),
         )
     return hostSpec
 }
diff --git a/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/HostSpec.kt b/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/HostSpec.kt
index 30a75896..8bfef464 100644
--- a/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/HostSpec.kt
+++ b/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/HostSpec.kt
@@ -24,6 +24,7 @@ package org.opendc.compute.topology.specs
 
 import org.opendc.simulator.compute.models.MachineModel
 import org.opendc.simulator.compute.power.PowerModel
+import org.opendc.simulator.engine.graph.distributionPolicies.FlowDistributorFactory.DistributionPolicy
 
 /**
  * Description of a physical host that will be simulated by OpenDC and host the virtual machines.
@@ -40,4 +41,6 @@ public data class HostSpec(
     val gpuPowerModel: PowerModel?,
     val embodiedCarbon: Double = 1000.0,
     val expectedLifetime: Double = 5.0,
+    val cpuDistributionPolicy: DistributionPolicy = DistributionPolicy.MAX_MIN_FAIRNESS,
+    val gpuDistributionPolicy: DistributionPolicy = DistributionPolicy.MAX_MIN_FAIRNESS,
 )
diff --git a/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/TopologySpecs.kt b/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/TopologySpecs.kt
index 62c3906a..f5c8ab31 100644
--- a/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/TopologySpecs.kt
+++ b/opendc-compute/opendc-compute-topology/src/main/kotlin/org/opendc/compute/topology/specs/TopologySpecs.kt
@@ -36,6 +36,7 @@ import org.opendc.simulator.compute.power.batteries.policy.RunningMeanBatteryPol
 import org.opendc.simulator.compute.power.batteries.policy.RunningMeanPlusBatteryPolicy
 import org.opendc.simulator.compute.power.batteries.policy.SingleThresholdBatteryPolicy
 import org.opendc.simulator.engine.engine.FlowEngine
+import org.opendc.simulator.engine.graph.distributionPolicies.FlowDistributorFactory.DistributionPolicy
 
 /**
  * Definition of a Topology modeled in the simulation.
@@ -80,6 +81,8 @@ public data class HostJSONSpec(
     val gpu: GPUJSONSpec? = null,
     val cpuPowerModel: PowerModelSpec = PowerModelSpec.DFLT,
     val gpuPowerModel: PowerModelSpec = PowerModelSpec.DFLT,
+    val cpuDistributionPolicy: DistributionPolicySpec = MaxMinFairnessDistributionPolicySpec(),
+    val gpuDistributionPolicy: DistributionPolicySpec = MaxMinFairnessDistributionPolicySpec(),
 )
 
 /**
@@ -160,6 +163,59 @@ public data class PowerModelSpec(
     }
 }
 
+@Serializable
+public sealed interface DistributionPolicySpec {
+    public val type: DistributionPolicy
+}
+
+@Serializable
+@SerialName("BEST_EFFORT")
+public data class BestEffortDistributionPolicySpec(
+    override val type: DistributionPolicy = DistributionPolicy.BEST_EFFORT,
+    val updateIntervalLength: Long = 1000L,
+) : DistributionPolicySpec
+
+@Serializable
+@SerialName("EQUAL_SHARE")
+public data class EqualShareDistributionPolicySpec(
+    override val type: DistributionPolicy = DistributionPolicy.EQUAL_SHARE,
+) : DistributionPolicySpec
+
+@Serializable
+@SerialName("FIRST_FIT")
+public data class FirstFitDistributionPolicySpec(
+    override val type: DistributionPolicy = DistributionPolicy.FIRST_FIT,
+) : DistributionPolicySpec
+
+@Serializable
+@SerialName("FIXED_SHARE")
+public data class FixedShareDistributionPolicySpec(
+    override val type: DistributionPolicy = DistributionPolicy.FIXED_SHARE,
+    val shareRatio: Double = 1.0,
+) : DistributionPolicySpec
+
+public fun DistributionPolicySpec.toDistributionPolicy(): DistributionPolicy {
+    return when (this) {
+        is BestEffortDistributionPolicySpec ->
+            DistributionPolicy.BEST_EFFORT.apply {
+                setProperty("updateIntervalLength", updateIntervalLength)
+            }
+        is EqualShareDistributionPolicySpec -> DistributionPolicy.EQUAL_SHARE
+        is FixedShareDistributionPolicySpec ->
+            DistributionPolicy.FIXED_SHARE.apply {
+                setProperty("shareRatio", shareRatio)
+            }
+        is FirstFitDistributionPolicySpec -> DistributionPolicy.FIRST_FIT
+        is MaxMinFairnessDistributionPolicySpec -> DistributionPolicy.MAX_MIN_FAIRNESS
+    }
+}
+
+@Serializable
+@SerialName("MAX_MIN_FAIRNESS")
+public data class MaxMinFairnessDistributionPolicySpec(
+    override val type: DistributionPolicy = DistributionPolicy.MAX_MIN_FAIRNESS,
+) : DistributionPolicySpec
+
 /**
  * Definition of a power source used for JSON input.
  *