Merge pull request #71 from atlarge-research/perf/workload

Convert to pull-based workload model

5 files changed, 17 insertions, 17 deletions

diff --git a/simulator/opendc-experiments/opendc-experiments-sc18/src/main/kotlin/org/opendc/experiments/sc18/TestExperiment.kt b/simulator/opendc-experiments/opendc-experiments-sc18/src/main/kotlin/org/opendc/experiments/sc18/TestExperiment.kt
index 9ad744f2..202df6df 100644
--- a/simulator/opendc-experiments/opendc-experiments-sc18/src/main/kotlin/org/opendc/experiments/sc18/TestExperiment.kt
+++ b/simulator/opendc-experiments/opendc-experiments-sc18/src/main/kotlin/org/opendc/experiments/sc18/TestExperiment.kt
@@ -32,6 +32,7 @@ import org.opendc.compute.simulator.SimVirtProvisioningService
 import org.opendc.compute.simulator.allocation.NumberOfActiveServersAllocationPolicy
 import org.opendc.format.environment.sc18.Sc18EnvironmentReader
 import org.opendc.format.trace.gwf.GwfTraceReader
+import org.opendc.simulator.compute.SimSpaceSharedHypervisorProvider
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.trace.core.EventTracer
 import org.opendc.workflows.service.StageWorkflowService
@@ -71,7 +72,7 @@ public fun main(args: Array<String>) {
         // Wait for the bare metal nodes to be spawned
         delay(10)
 
-        val provisioner = SimVirtProvisioningService(testScope, clock, bareMetal, NumberOfActiveServersAllocationPolicy(), tracer, schedulingQuantum = 1000)
+        val provisioner = SimVirtProvisioningService(testScope, clock, bareMetal, NumberOfActiveServersAllocationPolicy(), tracer, SimSpaceSharedHypervisorProvider(), schedulingQuantum = 1000)
 
         // Wait for the hypervisors to be spawned
         delay(10)
diff --git a/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/experiment/ExperimentHelpers.kt b/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/experiment/ExperimentHelpers.kt
index f939738d..1e01e892 100644
--- a/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/experiment/ExperimentHelpers.kt
+++ b/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/experiment/ExperimentHelpers.kt
@@ -48,6 +48,7 @@ import org.opendc.experiments.sc20.experiment.monitor.ExperimentMonitor
 import org.opendc.experiments.sc20.trace.Sc20StreamingParquetTraceReader
 import org.opendc.format.environment.EnvironmentReader
 import org.opendc.format.trace.TraceReader
+import org.opendc.simulator.compute.SimFairShareHypervisorProvider
 import org.opendc.simulator.compute.interference.PerformanceInterferenceModel
 import org.opendc.simulator.failures.CorrelatedFaultInjector
 import org.opendc.simulator.failures.FailureDomain
@@ -150,7 +151,7 @@ public suspend fun createProvisioner(
     // Wait for the bare metal nodes to be spawned
     delay(10)
 
-    val scheduler = SimVirtProvisioningService(coroutineScope, clock, bareMetalProvisioner, allocationPolicy, eventTracer)
+    val scheduler = SimVirtProvisioningService(coroutineScope, clock, bareMetalProvisioner, allocationPolicy, eventTracer, SimFairShareHypervisorProvider())
 
     // Wait for the hypervisors to be spawned
     delay(10)
diff --git a/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20RawParquetTraceReader.kt b/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20RawParquetTraceReader.kt
index 9bc1a58e..4a318df4 100644
--- a/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20RawParquetTraceReader.kt
+++ b/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20RawParquetTraceReader.kt
@@ -66,8 +66,6 @@ public class Sc20RawParquetTraceReader(private val path: File) {
                 val flops = record["flops"] as Long
 
                 val fragment = SimTraceWorkload.Fragment(
-                    tick,
-                    flops,
                     duration,
                     cpuUsage,
                     cores
diff --git a/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20StreamingParquetTraceReader.kt b/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20StreamingParquetTraceReader.kt
index edef276c..ba22ae15 100644
--- a/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20StreamingParquetTraceReader.kt
+++ b/simulator/opendc-experiments/opendc-experiments-sc20/src/main/kotlin/org/opendc/experiments/sc20/trace/Sc20StreamingParquetTraceReader.kt
@@ -92,7 +92,7 @@ public class Sc20StreamingParquetTraceReader(
     /**
      * A poisonous fragment.
      */
-    private val poison = Pair("\u0000", SimTraceWorkload.Fragment(0, 0, 0, 0.0, 0))
+    private val poison = Pair("\u0000", SimTraceWorkload.Fragment(0, 0.0, 0))
 
     /**
      * The thread to read the records in.
@@ -120,8 +120,6 @@ public class Sc20StreamingParquetTraceReader(
                 val flops = record["flops"] as Long
 
                 val fragment = SimTraceWorkload.Fragment(
-                    tick,
-                    flops,
                     duration,
                     cpuUsage,
                     cores
@@ -204,6 +202,7 @@ public class Sc20StreamingParquetTraceReader(
                 val externalBuffer = mutableListOf<SimTraceWorkload.Fragment>()
                 buffers.getOrPut(id) { mutableListOf() }.add(externalBuffer)
                 val fragments = sequence {
+                    var time = submissionTime
                     repeat@ while (true) {
                         if (externalBuffer.isEmpty()) {
                             if (hasNext) {
@@ -220,7 +219,8 @@ public class Sc20StreamingParquetTraceReader(
                         for (fragment in internalBuffer) {
                             yield(fragment)
 
-                            if (fragment.time >= endTime) {
+                            time += fragment.duration
+                            if (time >= endTime) {
                                 break@repeat
                             }
                         }
diff --git a/simulator/opendc-experiments/opendc-experiments-sc20/src/test/kotlin/org/opendc/experiments/sc20/Sc20IntegrationTest.kt b/simulator/opendc-experiments/opendc-experiments-sc20/src/test/kotlin/org/opendc/experiments/sc20/Sc20IntegrationTest.kt
index 2eedb636..c5ad345d 100644
--- a/simulator/opendc-experiments/opendc-experiments-sc20/src/test/kotlin/org/opendc/experiments/sc20/Sc20IntegrationTest.kt
+++ b/simulator/opendc-experiments/opendc-experiments-sc20/src/test/kotlin/org/opendc/experiments/sc20/Sc20IntegrationTest.kt
@@ -90,7 +90,7 @@ class Sc20IntegrationTest {
     fun tearDown() = testScope.cleanupTestCoroutines()
 
     @Test
-    fun smoke() {
+    fun testLarge() {
         val failures = false
         val seed = 0
         val chan = Channel<Unit>(Channel.CONFLATED)
@@ -148,15 +148,15 @@ class Sc20IntegrationTest {
         assertAll(
             { assertEquals(50, scheduler.submittedVms, "The trace contains 50 VMs") },
             { assertEquals(50, scheduler.finishedVms, "All VMs should finish after a run") },
-            { assertEquals(207480856422, monitor.totalRequestedBurst) },
-            { assertEquals(206510493178, monitor.totalGrantedBurst) },
-            { assertEquals(336120436, monitor.totalOvercommissionedBurst) },
+            { assertEquals(1684849230562, monitor.totalRequestedBurst) },
+            { assertEquals(447612683996, monitor.totalGrantedBurst) },
+            { assertEquals(1219535757406, monitor.totalOvercommissionedBurst) },
             { assertEquals(0, monitor.totalInterferedBurst) }
         )
     }
 
     @Test
-    fun small() {
+    fun testSmall() {
         val seed = 1
         val chan = Channel<Unit>(Channel.CONFLATED)
         val allocationPolicy = AvailableCoreMemoryAllocationPolicy()
@@ -195,10 +195,10 @@ class Sc20IntegrationTest {
 
         // Note that these values have been verified beforehand
         assertAll(
-            { assertEquals(96410877173, monitor.totalRequestedBurst) },
-            { assertEquals(96046583992, monitor.totalGrantedBurst) },
-            { assertEquals(19265632, monitor.totalOvercommissionedBurst) },
-            { assertEquals(0, monitor.totalInterferedBurst) }
+            { assertEquals(705128393966, monitor.totalRequestedBurst) { "Total requested work incorrect" } },
+            { assertEquals(173489747029, monitor.totalGrantedBurst) { "Total granted work incorrect" } },
+            { assertEquals(526858997740, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
+            { assertEquals(0, monitor.totalInterferedBurst) { "Total interfered work incorrect" } }
         )
     }