Merge pull request #16 from atlarge-research/feat/hpc-sampling

Finish HPC sampler

3 files changed, 54 insertions, 17 deletions

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 b8dfb1be..09a6ce40 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
@@ -203,6 +203,7 @@ public class MoreHpcPortfolio(parent: Experiment, id: Int) : Portfolio(parent, i
     )
 
     override val workloads = listOf(
+        Workload("solvinity", 0.0, samplingStrategy = SamplingStrategy.HPC),
         Workload("solvinity", 0.25, samplingStrategy = SamplingStrategy.HPC),
         Workload("solvinity", 0.5, samplingStrategy = SamplingStrategy.HPC),
         Workload("solvinity", 1.0, samplingStrategy = SamplingStrategy.HPC),
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 b24d6de1..a46bb3e6 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
@@ -109,6 +109,24 @@ fun sampleHpcWorkload(
         name.matches(pattern)
     }
 
+    val hpcSequence = generateSequence(0) { it + 1 }
+        .map { index ->
+            val res = mutableListOf<TraceEntry<VmWorkload>>()
+            hpc.mapTo(res) { sample(it, index) }
+            res.shuffle(random)
+            res
+        }
+        .flatten()
+
+    val nonHpcSequence = generateSequence(0) { it + 1 }
+        .map { index ->
+            val res = mutableListOf<TraceEntry<VmWorkload>>()
+            nonHpc.mapTo(res) { sample(it, index) }
+            res.shuffle(random)
+            res
+        }
+        .flatten()
+
     logger.debug { "Found ${hpc.size} HPC workloads and ${nonHpc.size} non-HPC workloads" }
 
     val totalLoad = if (workload is CompositeWorkload) {
@@ -117,45 +135,60 @@ fun sampleHpcWorkload(
         trace.sumByDouble { it.workload.image.tags.getValue("total-load") as Double }
     }
 
+    logger.debug { "Total trace load: $totalLoad" }
+    var hpcCount = 0
+    var hpcLoad = 0.0
+    var nonHpcCount = 0
+    var nonHpcLoad = 0.0
+
     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++)
-
+        for (entry in hpcSequence) {
             val entryLoad = entry.workload.image.tags.getValue("total-load") as Double
-            if ((currentLoad + entryLoad) / totalLoad > fraction || res.size > trace.size) {
+            if ((currentLoad + entryLoad) / totalLoad > fraction) {
                 break
             }
 
+            hpcLoad += entryLoad
+            hpcCount += 1
             currentLoad += entryLoad
             res += entry
         }
 
-        (nonHpc as MutableList<TraceEntry<VmWorkload>>).shuffle(random)
-        for (entry in nonHpc) {
+        for (entry in nonHpcSequence) {
             val entryLoad = entry.workload.image.tags.getValue("total-load") as Double
-            if ((currentLoad + entryLoad) / totalLoad > 1 || res.size > trace.size) {
+            if ((currentLoad + entryLoad) / totalLoad > 1) {
                 break
             }
 
+            nonHpcLoad += entryLoad
+            nonHpcCount += 1
             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)
-        }
+        hpcSequence
+            .take((fraction * trace.size).toInt())
+            .forEach { entry ->
+                hpcLoad += entry.workload.image.tags.getValue("total-load") as Double
+                hpcCount += 1
+                res.add(entry)
+            }
 
-        (nonHpc as MutableList<TraceEntry<VmWorkload>>).shuffle(random)
-        res.addAll(nonHpc.subList(0, ((1 - fraction) * trace.size).toInt()))
+        nonHpcSequence
+            .take(((1 - fraction) * trace.size).toInt())
+            .forEach { entry ->
+                nonHpcLoad += entry.workload.image.tags.getValue("total-load") as Double
+                nonHpcCount += 1
+                res.add(entry)
+            }
     }
 
+    logger.debug { "HPC $hpcCount (load $hpcLoad) and non-HPC $nonHpcCount (load $nonHpcLoad)" }
+    logger.debug { "Total sampled load: ${hpcLoad + nonHpcLoad}" }
     logger.info { "Sampled ${trace.size} VMs (fraction $fraction) into subset of ${res.size} VMs" }
 
     return res
@@ -168,13 +201,13 @@ 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.name,
         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")
+    val vmWorkload = entry.workload.copy(uid = id, image = image, name = entry.workload.name)
     return VmTraceEntry(vmWorkload, entry.submissionTime)
 }
 
diff --git a/simulator/opendc/opendc-experiments-sc20/src/main/resources/log4j2.xml b/simulator/opendc/opendc-experiments-sc20/src/main/resources/log4j2.xml
index f47a6da8..6906bfc3 100644
--- a/simulator/opendc/opendc-experiments-sc20/src/main/resources/log4j2.xml
+++ b/simulator/opendc/opendc-experiments-sc20/src/main/resources/log4j2.xml
@@ -39,6 +39,9 @@
         <Logger name="com.atlarge.opendc.experiments.sc20" level="info" additivity="false">
             <AppenderRef ref="Console"/>
         </Logger>
+        <Logger name="com.atlarge.opendc.experiments.sc20.trace" level="debug" additivity="false">
+            <AppenderRef ref="Console"/>
+        </Logger>
         <Logger name="org.apache.hadoop" level="warn" additivity="false">
             <AppenderRef ref="Console"/>
         </Logger>