Merge pull request #69 from atlarge-research/refactor/workflows-v1

Refactor workflow service to schedule tasks onto VMs

3 files changed, 23 insertions, 14 deletions

diff --git a/simulator/opendc-experiments/opendc-experiments-sc18/build.gradle.kts b/simulator/opendc-experiments/opendc-experiments-sc18/build.gradle.kts
index 9cf72f18..ee2295d9 100644
--- a/simulator/opendc-experiments/opendc-experiments-sc18/build.gradle.kts
+++ b/simulator/opendc-experiments/opendc-experiments-sc18/build.gradle.kts
@@ -37,6 +37,7 @@ dependencies {
     implementation(project(":opendc-format"))
     implementation(project(":opendc-workflows"))
     implementation(project(":opendc-simulator:opendc-simulator-core"))
+    implementation(project(":opendc-compute:opendc-compute-simulator"))
     implementation("com.fasterxml.jackson.module:jackson-module-kotlin:2.9.8") {
         exclude("org.jetbrains.kotlin", module = "kotlin-reflect")
     }
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 1221c7d3..9ad744f2 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
@@ -28,6 +28,8 @@ import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.test.TestCoroutineScope
 import org.opendc.compute.core.metal.service.ProvisioningService
+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.utils.DelayControllerClockAdapter
@@ -37,8 +39,6 @@ import org.opendc.workflows.service.WorkflowEvent
 import org.opendc.workflows.service.WorkflowSchedulerMode
 import org.opendc.workflows.service.stage.job.NullJobAdmissionPolicy
 import org.opendc.workflows.service.stage.job.SubmissionTimeJobOrderPolicy
-import org.opendc.workflows.service.stage.resource.FirstFitResourceSelectionPolicy
-import org.opendc.workflows.service.stage.resource.FunctionalResourceFilterPolicy
 import org.opendc.workflows.service.stage.task.NullTaskEligibilityPolicy
 import org.opendc.workflows.service.stage.task.SubmissionTimeTaskOrderPolicy
 import java.io.File
@@ -63,21 +63,29 @@ public fun main(args: Array<String>) {
     val tracer = EventTracer(clock)
 
     val schedulerAsync = testScope.async {
-        val environment = Sc18EnvironmentReader(object {}.javaClass.getResourceAsStream("/env/setup-test.json"))
-            .use { it.construct(this, clock) }
+        val environment = Sc18EnvironmentReader(object {}.javaClass.getResourceAsStream("/environment.json"))
+            .use { it.construct(testScope, clock) }
+
+        val bareMetal = environment.platforms[0].zones[0].services[ProvisioningService]
+
+        // Wait for the bare metal nodes to be spawned
+        delay(10)
+
+        val provisioner = SimVirtProvisioningService(testScope, clock, bareMetal, NumberOfActiveServersAllocationPolicy(), tracer, schedulingQuantum = 1000)
+
+        // Wait for the hypervisors to be spawned
+        delay(10)
 
         StageWorkflowService(
-            this,
+            testScope,
             clock,
             tracer,
-            environment.platforms[0].zones[0].services[ProvisioningService],
+            provisioner,
             mode = WorkflowSchedulerMode.Batch(100),
             jobAdmissionPolicy = NullJobAdmissionPolicy,
             jobOrderPolicy = SubmissionTimeJobOrderPolicy(),
             taskEligibilityPolicy = NullTaskEligibilityPolicy,
             taskOrderPolicy = SubmissionTimeTaskOrderPolicy(),
-            resourceFilterPolicy = FunctionalResourceFilterPolicy,
-            resourceSelectionPolicy = FirstFitResourceSelectionPolicy
         )
     }
 
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 72a2484a..2eedb636 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
@@ -148,9 +148,9 @@ class Sc20IntegrationTest {
         assertAll(
             { assertEquals(50, scheduler.submittedVms, "The trace contains 50 VMs") },
             { assertEquals(50, scheduler.finishedVms, "All VMs should finish after a run") },
-            { assertEquals(207379117949, monitor.totalRequestedBurst) },
-            { assertEquals(203388071813, monitor.totalGrantedBurst) },
-            { assertEquals(3991046136, monitor.totalOvercommissionedBurst) },
+            { assertEquals(207480856422, monitor.totalRequestedBurst) },
+            { assertEquals(206510493178, monitor.totalGrantedBurst) },
+            { assertEquals(336120436, monitor.totalOvercommissionedBurst) },
             { assertEquals(0, monitor.totalInterferedBurst) }
         )
     }
@@ -195,9 +195,9 @@ class Sc20IntegrationTest {
 
         // Note that these values have been verified beforehand
         assertAll(
-            { assertEquals(96344114723, monitor.totalRequestedBurst) },
-            { assertEquals(96324378235, monitor.totalGrantedBurst) },
-            { assertEquals(19736424, monitor.totalOvercommissionedBurst) },
+            { assertEquals(96410877173, monitor.totalRequestedBurst) },
+            { assertEquals(96046583992, monitor.totalGrantedBurst) },
+            { assertEquals(19265632, monitor.totalOvercommissionedBurst) },
             { assertEquals(0, monitor.totalInterferedBurst) }
         )
     }