Merge pull request #11 from atlarge-research/feat/hpc-sampler

Add HPC workload sampler and portfolio

4 files changed, 136 insertions, 17 deletions

diff --git a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Main.kt b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Main.kt
index faa68e34..4781f335 100644
--- a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Main.kt
+++ b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Main.kt
@@ -24,14 +24,7 @@
 
 package com.atlarge.opendc.experiments.sc20
 
-import com.atlarge.opendc.experiments.sc20.experiment.CompositeWorkloadPortfolio
-import com.atlarge.opendc.experiments.sc20.experiment.Experiment
-import com.atlarge.opendc.experiments.sc20.experiment.HorVerPortfolio
-import com.atlarge.opendc.experiments.sc20.experiment.MoreVelocityPortfolio
-import com.atlarge.opendc.experiments.sc20.experiment.OperationalPhenomenaPortfolio
-import com.atlarge.opendc.experiments.sc20.experiment.Portfolio
-import com.atlarge.opendc.experiments.sc20.experiment.ReplayPortfolio
-import com.atlarge.opendc.experiments.sc20.experiment.TestPortfolio
+import com.atlarge.opendc.experiments.sc20.experiment.*
 import com.atlarge.opendc.experiments.sc20.reporter.ConsoleExperimentReporter
 import com.atlarge.opendc.experiments.sc20.runner.ExperimentDescriptor
 import com.atlarge.opendc.experiments.sc20.runner.execution.ThreadPoolExperimentScheduler
@@ -104,6 +97,7 @@ class ExperimentCli : CliktCommand(name = "sc20-experiment") {
             "operational-phenomena" to { experiment, i -> OperationalPhenomenaPortfolio(experiment, i) },
             "replay" to { experiment, i -> ReplayPortfolio(experiment, i) },
             "test" to { experiment, i -> TestPortfolio(experiment, i) },
+            "more-hpc" to { experiment, i -> MoreHpcPortfolio(experiment, i) },
             ignoreCase = true
         )
         .multiple()
diff --git a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/Portfolios.kt b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/Portfolios.kt
index 4d8e2f1d..b8dfb1be 100644
--- a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/Portfolios.kt
+++ b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/Portfolios.kt
@@ -24,10 +24,7 @@
 
 package com.atlarge.opendc.experiments.sc20.experiment
 
-import com.atlarge.opendc.experiments.sc20.experiment.model.CompositeWorkload
-import com.atlarge.opendc.experiments.sc20.experiment.model.OperationalPhenomena
-import com.atlarge.opendc.experiments.sc20.experiment.model.Topology
-import com.atlarge.opendc.experiments.sc20.experiment.model.Workload
+import com.atlarge.opendc.experiments.sc20.experiment.model.*
 
 public class HorVerPortfolio(parent: Experiment, id: Int) : Portfolio(parent, id, "horizontal_vs_vertical") {
     override val topologies = listOf(
@@ -196,3 +193,29 @@ public class TestPortfolio(parent: Experiment, id: Int) : Portfolio(parent, id,
 
     override val allocationPolicies: List<String> = listOf("active-servers")
 }
+
+public class MoreHpcPortfolio(parent: Experiment, id: Int) : Portfolio(parent, id, "more_hpc") {
+    override val topologies = listOf(
+        Topology("base"),
+        Topology("exp-vol-hor-hom"),
+        Topology("exp-vol-ver-hom"),
+        Topology("exp-vel-ver-hom")
+    )
+
+    override val workloads = listOf(
+        Workload("solvinity", 0.25, samplingStrategy = SamplingStrategy.HPC),
+        Workload("solvinity", 0.5, samplingStrategy = SamplingStrategy.HPC),
+        Workload("solvinity", 1.0, samplingStrategy = SamplingStrategy.HPC),
+        Workload("solvinity", 0.25, samplingStrategy = SamplingStrategy.HPC_LOAD),
+        Workload("solvinity", 0.5, samplingStrategy = SamplingStrategy.HPC_LOAD),
+        Workload("solvinity", 1.0, samplingStrategy = SamplingStrategy.HPC_LOAD)
+    )
+
+    override val operationalPhenomenas = listOf(
+        OperationalPhenomena(failureFrequency = 24.0 * 7, hasInterference = true)
+    )
+
+    override val allocationPolicies = listOf(
+        "active-servers"
+    )
+}
diff --git a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/model/Workload.kt b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/model/Workload.kt
index cc3c448a..d75ca6f9 100644
--- a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/model/Workload.kt
+++ b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/experiment/model/Workload.kt
@@ -24,10 +24,16 @@
 
 package com.atlarge.opendc.experiments.sc20.experiment.model
 
+enum class SamplingStrategy {
+    REGULAR,
+    HPC,
+    HPC_LOAD
+}
+
 /**
  * A workload that is considered for a scenario.
  */
-public open class Workload(open val name: String, val fraction: Double)
+public open class Workload(open val name: String, val fraction: Double, val samplingStrategy: SamplingStrategy = SamplingStrategy.REGULAR)
 
 /**
  * A workload that is composed of multiple workloads.
diff --git a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/WorkloadSampler.kt b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/WorkloadSampler.kt
index f2a0e627..b24d6de1 100644
--- a/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/WorkloadSampler.kt
+++ b/simulator/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/WorkloadSampler.kt
@@ -24,10 +24,13 @@
 
 package com.atlarge.opendc.experiments.sc20.trace
 
+import com.atlarge.opendc.compute.core.image.VmImage
 import com.atlarge.opendc.compute.core.workload.VmWorkload
 import com.atlarge.opendc.experiments.sc20.experiment.model.CompositeWorkload
+import com.atlarge.opendc.experiments.sc20.experiment.model.SamplingStrategy
 import com.atlarge.opendc.experiments.sc20.experiment.model.Workload
 import com.atlarge.opendc.format.trace.TraceEntry
+import java.util.*
 import kotlin.random.Random
 import mu.KotlinLogging
 
@@ -42,10 +45,14 @@ fun sampleWorkload(
     subWorkload: Workload,
     seed: Int
 ): List<TraceEntry<VmWorkload>> {
-    return if (workload is CompositeWorkload) {
-        sampleRegularWorkload(trace, workload, subWorkload, seed)
-    } else {
-        sampleRegularWorkload(trace, workload, workload, seed)
+    return when {
+        workload is CompositeWorkload -> sampleRegularWorkload(trace, workload, subWorkload, seed)
+        workload.samplingStrategy == SamplingStrategy.HPC ->
+            sampleHpcWorkload(trace, workload, seed, sampleOnLoad = false)
+        workload.samplingStrategy == SamplingStrategy.HPC_LOAD ->
+            sampleHpcWorkload(trace, workload, seed, sampleOnLoad = true)
+        else ->
+            sampleRegularWorkload(trace, workload, workload, seed)
     }
 }
 
@@ -83,3 +90,92 @@ fun sampleRegularWorkload(
 
     return res
 }
+
+/**
+ * Sample a HPC workload.
+ */
+fun sampleHpcWorkload(
+    trace: List<TraceEntry<VmWorkload>>,
+    workload: Workload,
+    seed: Int,
+    sampleOnLoad: Boolean
+): List<TraceEntry<VmWorkload>> {
+    val pattern = Regex("^vm__workload__(ComputeNode|cn).*")
+    val random = Random(seed)
+
+    val fraction = workload.fraction
+    val (hpc, nonHpc) = trace.partition { entry ->
+        val name = entry.workload.image.name
+        name.matches(pattern)
+    }
+
+    logger.debug { "Found ${hpc.size} HPC workloads and ${nonHpc.size} non-HPC workloads" }
+
+    val totalLoad = if (workload is CompositeWorkload) {
+        workload.totalLoad
+    } else {
+        trace.sumByDouble { it.workload.image.tags.getValue("total-load") as Double }
+    }
+
+    val res = mutableListOf<TraceEntry<VmWorkload>>()
+
+    if (sampleOnLoad) {
+        var currentLoad = 0.0
+        var i = 0
+        while (true) {
+            // Sample random HPC entry with replacement
+            val entry = sample(hpc.random(random), i++)
+
+            val entryLoad = entry.workload.image.tags.getValue("total-load") as Double
+            if ((currentLoad + entryLoad) / totalLoad > fraction || res.size > trace.size) {
+                break
+            }
+
+            currentLoad += entryLoad
+            res += entry
+        }
+
+        (nonHpc as MutableList<TraceEntry<VmWorkload>>).shuffle(random)
+        for (entry in nonHpc) {
+            val entryLoad = entry.workload.image.tags.getValue("total-load") as Double
+            if ((currentLoad + entryLoad) / totalLoad > 1 || res.size > trace.size) {
+                break
+            }
+
+            currentLoad += entryLoad
+            res += entry
+        }
+    } else {
+        repeat((fraction * trace.size).toInt()) { i ->
+            // Sample random HPC entry with replacement
+            val entry = sample(hpc.random(random), i)
+            res.add(entry)
+        }
+
+        (nonHpc as MutableList<TraceEntry<VmWorkload>>).shuffle(random)
+        res.addAll(nonHpc.subList(0, ((1 - fraction) * trace.size).toInt()))
+    }
+
+    logger.info { "Sampled ${trace.size} VMs (fraction $fraction) into subset of ${res.size} VMs" }
+
+    return res
+}
+
+/**
+ * Sample a random trace entry.
+ */
+private fun sample(entry: TraceEntry<VmWorkload>, i: Int): TraceEntry<VmWorkload> {
+    val id = UUID.nameUUIDFromBytes("${entry.workload.image.uid}-$i".toByteArray())
+    val image = VmImage(
+        id,
+        entry.workload.image.name + "-$i",
+        entry.workload.image.tags,
+        entry.workload.image.flopsHistory,
+        entry.workload.image.maxCores,
+        entry.workload.image.requiredMemory
+    )
+    val vmWorkload = entry.workload.copy(uid = id, image = image, name = entry.workload.name + "-$i")
+    return VmTraceEntry(vmWorkload, entry.submissionTime)
+}
+
+private class VmTraceEntry(override val workload: VmWorkload, override val submissionTime: Long) : TraceEntry<VmWorkload>