diff options
26 files changed, 810 insertions, 163 deletions
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 807c6d26..71fb15a8 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -22,7 +22,7 @@ jobs: - name: Checkout repository uses: actions/checkout@v4 - name: Validate Gradle wrapper - uses: gradle/wrapper-validation-action@v1 + uses: gradle/wrapper-validation-action@v3 - name: Set up JDK uses: actions/setup-java@v4 with: diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index e2b09550..660b7899 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -90,7 +90,7 @@ jobs: - name: Checkout repository uses: actions/checkout@v4 - name: Validate Gradle wrapper - uses: gradle/wrapper-validation-action@v1 + uses: gradle/wrapper-validation-action@v3 - name: Set up JDK uses: actions/setup-java@v3 with: diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5b3357f4..3079d25a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -13,7 +13,7 @@ jobs: - name: Checkout repository uses: actions/checkout@v4 - name: Validate Gradle wrapper - uses: gradle/wrapper-validation-action@v1 + uses: gradle/wrapper-validation-action@v3 - name: Set up JDK uses: actions/setup-java@v3 with: diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt index ee2cb319..7f5f09eb 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt @@ -31,8 +31,9 @@ import org.opendc.compute.simulator.telemetry.GuestSystemStats import org.opendc.simulator.compute.machine.SimMachine import org.opendc.simulator.compute.machine.VirtualMachine import org.opendc.simulator.compute.workload.ChainWorkload -import org.opendc.simulator.compute.workload.TraceFragment -import org.opendc.simulator.compute.workload.TraceWorkload +import org.opendc.simulator.compute.workload.trace.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceWorkload +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling import java.time.Duration import java.time.Instant import java.time.InstantSource @@ -93,6 +94,8 @@ public class Guest( onStart() + val scalingPolicy = NoDelayScaling() + val bootworkload = TraceWorkload( ArrayList( @@ -107,6 +110,7 @@ public class Guest( 0, 0, 0.0, + scalingPolicy, ) if (task.workload is TraceWorkload) { diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt index 655bacb9..2b8b589f 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt @@ -23,7 +23,9 @@ package org.opendc.compute.workload import mu.KotlinLogging -import org.opendc.simulator.compute.workload.TraceWorkload +import org.opendc.simulator.compute.workload.trace.TraceWorkload +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling +import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy import org.opendc.trace.Trace import org.opendc.trace.conv.TABLE_RESOURCES import org.opendc.trace.conv.TABLE_RESOURCE_STATES @@ -53,6 +55,7 @@ public class ComputeWorkloadLoader( private val checkpointInterval: Long = 0L, private val checkpointDuration: Long = 0L, private val checkpointIntervalScaling: Double = 1.0, + private val scalingPolicy: ScalingPolicy = NoDelayScaling(), ) : WorkloadLoader(subMissionTime) { /** * The logger for this instance. @@ -84,7 +87,10 @@ public class ComputeWorkloadLoader( val cores = reader.getInt(coresCol) val cpuUsage = reader.getDouble(usageCol) - val builder = fragments.computeIfAbsent(id) { Builder(checkpointInterval, checkpointDuration, checkpointIntervalScaling) } + val builder = + fragments.computeIfAbsent( + id, + ) { Builder(checkpointInterval, checkpointDuration, checkpointIntervalScaling, scalingPolicy) } builder.add(durationMs, cpuUsage, cores) } @@ -178,7 +184,12 @@ public class ComputeWorkloadLoader( /** * A builder for a VM trace. */ - private class Builder(checkpointInterval: Long, checkpointDuration: Long, checkpointIntervalScaling: Double) { + private class Builder( + checkpointInterval: Long, + checkpointDuration: Long, + checkpointIntervalScaling: Double, + scalingPolicy: ScalingPolicy, + ) { /** * The total load of the trace. */ @@ -187,7 +198,7 @@ public class ComputeWorkloadLoader( /** * The internal builder for the trace. */ - private val builder = TraceWorkload.builder(checkpointInterval, checkpointDuration, checkpointIntervalScaling) + private val builder = TraceWorkload.builder(checkpointInterval, checkpointDuration, checkpointIntervalScaling, scalingPolicy) /** * Add a fragment to the trace. diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt index 60be9299..7a5089b9 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/Task.kt @@ -22,7 +22,7 @@ package org.opendc.compute.workload -import org.opendc.simulator.compute.workload.TraceWorkload +import org.opendc.simulator.compute.workload.trace.TraceWorkload import java.time.Instant import java.util.UUID diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt index 87c7abe9..cf40d88d 100644 --- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt +++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/experiment/specs/WorkloadSpec.kt @@ -25,6 +25,9 @@ package org.opendc.experiments.base.experiment.specs import kotlinx.serialization.Serializable import org.opendc.compute.workload.ComputeWorkloadLoader import org.opendc.compute.workload.WorkloadLoader +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling +import org.opendc.simulator.compute.workload.trace.scaling.PerfectScaling +import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy import java.io.File /** @@ -41,6 +44,7 @@ public data class WorkloadSpec( val type: WorkloadTypes, val sampleFraction: Double = 1.0, val submissionTime: String? = null, + val scalingPolicy: ScalingPolicyEnum = ScalingPolicyEnum.NoDelay, ) { public val name: String = File(pathToFile).nameWithoutExtension @@ -56,11 +60,6 @@ public data class WorkloadSpec( * @constructor Create empty Workload types */ public enum class WorkloadTypes { - /** - * Compute workload - * - * @constructor Create empty Compute workload - */ ComputeWorkload, } @@ -74,6 +73,7 @@ public fun getWorkloadLoader( checkpointInterval: Long, checkpointDuration: Long, checkpointIntervalScaling: Double, + scalingPolicy: ScalingPolicy, ): WorkloadLoader { return when (type) { WorkloadTypes.ComputeWorkload -> @@ -83,6 +83,19 @@ public fun getWorkloadLoader( checkpointInterval, checkpointDuration, checkpointIntervalScaling, + scalingPolicy, ) } } + +public enum class ScalingPolicyEnum { + NoDelay, + Perfect, +} + +public fun getScalingPolicy(scalingPolicyEnum: ScalingPolicyEnum): ScalingPolicy { + return when (scalingPolicyEnum) { + ScalingPolicyEnum.NoDelay -> NoDelayScaling() + ScalingPolicyEnum.Perfect -> PerfectScaling() + } +} diff --git a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt index c9c2729d..56278bf2 100644 --- a/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt +++ b/opendc-experiments/opendc-experiments-base/src/main/kotlin/org/opendc/experiments/base/runner/ScenarioRunner.kt @@ -33,6 +33,7 @@ import org.opendc.compute.simulator.service.ComputeService import org.opendc.compute.simulator.telemetry.parquet.ParquetComputeMonitor import org.opendc.compute.topology.clusterTopology import org.opendc.experiments.base.experiment.Scenario +import org.opendc.experiments.base.experiment.specs.getScalingPolicy import org.opendc.experiments.base.experiment.specs.getWorkloadLoader import org.opendc.simulator.kotlin.runSimulation import java.io.File @@ -80,6 +81,8 @@ public fun runScenario( val checkpointDuration = scenario.checkpointModelSpec?.checkpointDuration ?: 0L val checkpointIntervalScaling = scenario.checkpointModelSpec?.checkpointIntervalScaling ?: 1.0 + val scalingPolicy = getScalingPolicy(scenario.workloadSpec.scalingPolicy) + val workloadLoader = getWorkloadLoader( scenario.workloadSpec.type, @@ -88,6 +91,7 @@ public fun runScenario( checkpointInterval, checkpointDuration, checkpointIntervalScaling, + scalingPolicy, ) val workload = workloadLoader.sampleByLoad(scenario.workloadSpec.sampleFraction) diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt index 091f506a..895eee92 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/CarbonTest.kt @@ -26,7 +26,7 @@ import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertAll import org.opendc.compute.workload.Task -import org.opendc.simulator.compute.workload.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceFragment import java.util.ArrayList /** diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentTest.kt index 56850558..e271fce7 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/ExperimentTest.kt @@ -26,7 +26,7 @@ import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertAll import org.opendc.compute.workload.Task -import org.opendc.simulator.compute.workload.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceFragment import java.util.ArrayList /** diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt index 90737ab6..3231f533 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FailuresAndCheckpointingTest.kt @@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertAll import org.opendc.compute.workload.Task import org.opendc.experiments.base.experiment.specs.TraceBasedFailureModelSpec -import org.opendc.simulator.compute.workload.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceFragment import java.util.ArrayList /** @@ -225,15 +225,12 @@ class FailuresAndCheckpointingTest { } /** - * Failure test 1: Single Task with checkpointing + * Checkpointing test 1: Single Task with checkpointing * In this test, a single task is scheduled that is interrupted by a failure after 5 min. - * Because there is no checkpointing, the full task has to be rerun. - * - * This means the final runtime is 20 minutes + * The system is using checkpointing, taking snapshots every minute. * - * When the task is running, it is using 50% of the cpu. - * This means that half of the time is active, and half is idle. - * When the task is failed, all time is idle. + * This means that after failure, only 6 minutes of the task is left. + * However, taking a snapshot takes 1 second, which means 9 seconds have to be added to the total runtime. */ @Test fun testCheckpoints1() { @@ -256,22 +253,20 @@ class FailuresAndCheckpointingTest { assertAll( { assertEquals((10 * 60000) + (9 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } }, - { assertEquals(((10 * 30000)).toLong(), monitor.hostIdleTimes["H01"]?.sum()) { "Idle time incorrect" } }, - { assertEquals(((10 * 30000) + (9 * 1000)).toLong(), monitor.hostActiveTimes["H01"]?.sum()) { "Active time incorrect" } }, - { assertEquals((10 * 60 * 150.0) + (9 * 200.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } }, + { assertEquals((10 * 60 * 150.0) + (9 * 150.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } }, ) } /** - * Failure test 2: Single Task with scaling checkpointing + * Checkpointing test 2: Single Task with checkpointing, higher cpu demand * In this test, a single task is scheduled that is interrupted by a failure after 5 min. - * Because there is no checkpointing, the full task has to be rerun. + * The system is using checkpointing, taking snapshots every minute. * - * This means the final runtime is 20 minutes + * This means that after failure, only 16 minutes of the task is left. + * However, taking a snapshot takes 1 second, which means 19 seconds have to be added to the total runtime. * - * When the task is running, it is using 50% of the cpu. - * This means that half of the time is active, and half is idle. - * When the task is failed, all time is idle. + * This is similar to the previous test, but the cpu demand of taking a snapshot is higher. + * The cpu demand of taking a snapshot is as high as the highest fragment */ @Test fun testCheckpoints2() { @@ -281,6 +276,86 @@ class FailuresAndCheckpointingTest { name = "0", fragments = arrayListOf( + TraceFragment(10 * 60 * 1000, 2000.0, 1), + TraceFragment(10 * 60 * 1000, 1000.0, 1), + ), + checkpointInterval = 60 * 1000L, + checkpointDuration = 1000L, + ), + ) + + val topology = createTopology("single_1_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals((20 * 60000) + (19 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + (10 * 60 * 200.0) + (10 * 60 * 150.0) + (19 * 200.0), + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect energy usage" } + }, + ) + } + + /** + * Checkpointing test 3: Single Task with checkpointing, higher cpu demand + * In this test, a single task is scheduled that is interrupted by a failure after 5 min. + * The system is using checkpointing, taking snapshots every minute. + * + * This means that after failure, only 16 minutes of the task is left. + * However, taking a snapshot takes 1 second, which means 19 seconds have to be added to the total runtime. + * + * This is similar to the previous test, but the fragments are reversed + * + */ + @Test + fun testCheckpoints3() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0, 1), + TraceFragment(10 * 60 * 1000, 2000.0, 1), + ), + checkpointInterval = 60 * 1000L, + checkpointDuration = 1000L, + ), + ) + + val topology = createTopology("single_1_2000.json") + + val monitor = runTest(topology, workload) + + assertAll( + { assertEquals((20 * 60000) + (19 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } }, + { + assertEquals( + (10 * 60 * 200.0) + (10 * 60 * 150.0) + (19 * 200.0), + monitor.hostEnergyUsages["H01"]?.sum(), + ) { "Incorrect energy usage" } + }, + ) + } + + /** + * Checkpointing test 4: Single Task with scaling checkpointing + * In this test, checkpointing is used, with a scaling factor of 1.5 + * + * This means that the interval between checkpoints starts at 1 min, but is multiplied by 1.5 every snapshot. + * + */ + @Test + fun testCheckpoints4() { + val workload: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( TraceFragment(10 * 60 * 1000, 1000.0, 1), ), checkpointInterval = 60 * 1000L, @@ -295,20 +370,18 @@ class FailuresAndCheckpointingTest { assertAll( { assertEquals((10 * 60000) + (4 * 1000), monitor.maxTimestamp) { "Total runtime incorrect" } }, - { assertEquals(((10 * 30000)).toLong(), monitor.hostIdleTimes["H01"]?.sum()) { "Idle time incorrect" } }, - { assertEquals(((10 * 30000) + (4 * 1000)).toLong(), monitor.hostActiveTimes["H01"]?.sum()) { "Active time incorrect" } }, - { assertEquals((10 * 60 * 150.0) + (4 * 200.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } }, + { assertEquals((10 * 60 * 150.0) + (4 * 150.0), monitor.hostEnergyUsages["H01"]?.sum()) { "Incorrect energy usage" } }, ) } /** - * Checkpoint test 3: Single Task, single failure with checkpointing + * Checkpointing test 5: Single Task, single failure with checkpointing * In this test, a single task is scheduled that is interrupted by a failure after 5 min. * Because there is no checkpointing, the full task has to be rerun. * */ @Test - fun testCheckpoints3() { + fun testCheckpoints5() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -336,23 +409,7 @@ class FailuresAndCheckpointingTest { { assertEquals((960 * 1000) + 5000, monitor.maxTimestamp) { "Total runtime incorrect" } }, { assertEquals( - ((300 * 1000) + (296 * 500) + (360 * 500)).toLong(), - monitor.hostIdleTimes["H01"]?.sum(), - ) { "Idle time incorrect" } - }, - { - assertEquals( - ((296 * 500) + 4000 + (360 * 500) + 5000).toLong(), - monitor.hostActiveTimes["H01"]?.sum(), - ) { "Active time incorrect" } - }, - { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(0)) { "Incorrect energy usage" } }, - { assertEquals(6000.0, monitor.hostEnergyUsages["H01"]?.get(5)) { "Incorrect energy usage" } }, - { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(10)) { "Incorrect energy usage" } }, - { - assertEquals( - (296 * 150.0) + (4 * 200.0) + (300 * 100.0) + - (360 * 150.0) + (5 * 200.0), + (665 * 150.0) + (300 * 100.0), monitor.hostEnergyUsages["H01"]?.sum(), ) { "Incorrect energy usage" } }, @@ -360,13 +417,13 @@ class FailuresAndCheckpointingTest { } /** - * Checkpoint test 4: Single Task, repeated failure with checkpointing + * Checkpointing test 6: Single Task, repeated failure with checkpointing * In this test, a single task is scheduled that is interrupted by a failure after 5 min. * Because there is no checkpointing, the full task has to be rerun. * */ @Test - fun testCheckpoints4() { + fun testCheckpoints6() { val workload: ArrayList<Task> = arrayListOf( createTestTask( @@ -394,23 +451,7 @@ class FailuresAndCheckpointingTest { { assertEquals((22 * 60000) + 1000, monitor.maxTimestamp) { "Total runtime incorrect" } }, { assertEquals( - ((10 * 60000) + (2 * 296 * 500) + (120 * 500)).toLong(), - monitor.hostIdleTimes["H01"]?.sum(), - ) { "Idle time incorrect" } - }, - { - assertEquals( - ((2 * 296 * 500) + 8000 + (120 * 500) + 1000).toLong(), - monitor.hostActiveTimes["H01"]?.sum(), - ) { "Active time incorrect" } - }, - { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(0)) { "Incorrect energy usage" } }, - { assertEquals(6000.0, monitor.hostEnergyUsages["H01"]?.get(5)) { "Incorrect energy usage" } }, - { assertEquals(9000.0, monitor.hostEnergyUsages["H01"]?.get(10)) { "Incorrect energy usage" } }, - { - assertEquals( - (2 * 296 * 150.0) + (8 * 200.0) + (600 * 100.0) + - (120 * 150.0) + (200.0), + (300 * 150.0) + (300 * 100.0) + (300 * 150.0) + (300 * 100.0) + (121 * 150.0), monitor.hostEnergyUsages["H01"]?.sum(), ) { "Incorrect energy usage" } }, diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt index 4a7c9341..3d733360 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FlowDistributorTest.kt @@ -26,7 +26,7 @@ import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertAll import org.opendc.compute.workload.Task -import org.opendc.simulator.compute.workload.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceFragment import java.util.ArrayList /** diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FragmentScalingTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FragmentScalingTest.kt new file mode 100644 index 00000000..b0aa3555 --- /dev/null +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/FragmentScalingTest.kt @@ -0,0 +1,337 @@ +/* + * Copyright (c) 2020 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.experiments.base + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertAll +import org.opendc.compute.workload.Task +import org.opendc.simulator.compute.workload.trace.TraceFragment +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling +import org.opendc.simulator.compute.workload.trace.scaling.PerfectScaling +import java.util.ArrayList + +/** + * Testing suite containing tests that specifically test the scaling of trace fragments + */ +class FragmentScalingTest { + /** + * Scaling test 1: A single fitting task + * In this test, a single task is scheduled that should fit the system. + * This means nothing will be delayed regardless of the scaling policy + */ + @Test + fun testScaling1() { + val workloadNoDelay: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 2000.0, 1), + TraceFragment(10 * 60 * 1000, 1000.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + ) + + val workloadPerfect: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 2000.0, 1), + TraceFragment(10 * 60 * 1000, 1000.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + ) + val topology = createTopology("single_1_2000.json") + + val monitorNoDelay = runTest(topology, workloadNoDelay) + val monitorPerfect = runTest(topology, workloadPerfect) + + assertAll( + { assertEquals(1200000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(1200000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(2000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(2000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(2000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } }, + ) + } + + /** + * Scaling test 2: A single task with a single non-fitting fragment + * In this test, a single task is scheduled that should not fit. + * This means the Task is getting only 2000 Mhz while it was demanding 4000 Mhz + * + * For the NoDelay scaling policy, the task should take the planned 10 min. + * For the Perfect scaling policy, the task should be slowed down by 50% resulting in a runtime of 20 min. + */ + @Test + fun testScaling2() { + val workloadNoDelay: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 4000.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + ) + + val workloadPerfect: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 4000.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + ) + val topology = createTopology("single_1_2000.json") + + val monitorNoDelay = runTest(topology, workloadNoDelay) + val monitorPerfect = runTest(topology, workloadPerfect) + + assertAll( + { assertEquals(600000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(1200000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(4000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(4000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(2000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + ) + } + + /** + * Scaling test 3: A single task that switches between fitting and not fitting + * In this test, a single task is scheduled has one fragment that does not fit + * This means the second fragment is getting only 2000 Mhz while it was demanding 4000 Mhz + * + * For the NoDelay scaling policy, the task should take the planned 30 min. + * For the Perfect scaling policy, the second fragment should be slowed down by 50% resulting in a runtime of 20 min, + * and a total runtime of 40 min. + */ + @Test + fun testScaling3() { + val workloadNoDelay: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0, 1), + TraceFragment(10 * 60 * 1000, 4000.0, 1), + TraceFragment(10 * 60 * 1000, 1500.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + ) + + val workloadPerfect: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0, 1), + TraceFragment(10 * 60 * 1000, 4000.0, 1), + TraceFragment(10 * 60 * 1000, 1500.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + ) + val topology = createTopology("single_1_2000.json") + + val monitorNoDelay = runTest(topology, workloadNoDelay) + val monitorPerfect = runTest(topology, workloadPerfect) + + assertAll( + { assertEquals(1800000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(2400000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(4000.0, monitorNoDelay.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(4000.0, monitorPerfect.taskCpuDemands["0"]?.get(9)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(2000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(9)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(1500.0, monitorNoDelay.taskCpuDemands["0"]?.get(19)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(4000.0, monitorPerfect.taskCpuDemands["0"]?.get(19)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(1500.0, monitorNoDelay.taskCpuSupplied["0"]?.get(19)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(2000.0, monitorPerfect.taskCpuSupplied["0"]?.get(19)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(1500.0, monitorPerfect.taskCpuDemands["0"]?.get(29)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(1500.0, monitorPerfect.taskCpuSupplied["0"]?.get(29)) { "The cpu supplied to task 0 is incorrect" } }, + ) + } + + /** + * Scaling test 4: Two tasks, that both fit + * In this test, two tasks are scheduled that both fit + * + * For both scaling policies, the tasks should run without delay. + */ + @Test + fun testScaling4() { + val workloadNoDelay: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + createTestTask( + name = "1", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 3000.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + ) + + val workloadPerfect: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 1000.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + createTestTask( + name = "1", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 3000.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + ) + val topology = createTopology("single_2_2000.json") + + val monitorNoDelay = runTest(topology, workloadNoDelay) + val monitorPerfect = runTest(topology, workloadPerfect) + + assertAll( + { assertEquals(600000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(600000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } }, + { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(3000.0, monitorNoDelay.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } }, + { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, + { assertEquals(3000.0, monitorPerfect.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } }, + { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(3000.0, monitorNoDelay.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } }, + { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, + { assertEquals(3000.0, monitorPerfect.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } }, + ) + } + + /** + * Scaling test 5: Two tasks, that don't fit together + * In this test, two tasks are scheduled that do not fit together + * This means the Task_1 is getting only 2000 Mhz while it was demanding 4000 Mhz + * + * For the NoDelay scaling policy, the tasks should complete in 10 min + * For the Perfect scaling policy, task_1 is delayed while task_0 is still going. + * In the first 10 min (while Task_0 is still running), Task_1 is running at 50%. + * This means that after Task_0 is done, Task_1 still needs to run for 5 minutes, making the total runtime 15 min. + */ + @Test + fun testScaling5() { + val workloadNoDelay: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 2000.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + createTestTask( + name = "1", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 4000.0, 1), + ), + scalingPolicy = NoDelayScaling(), + ), + ) + + val workloadPerfect: ArrayList<Task> = + arrayListOf( + createTestTask( + name = "0", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 2000.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + createTestTask( + name = "1", + fragments = + arrayListOf( + TraceFragment(10 * 60 * 1000, 4000.0, 1), + ), + scalingPolicy = PerfectScaling(), + ), + ) + val topology = createTopology("single_2_2000.json") + +// val monitorNoDelay = runTest(topology, workloadNoDelay) + val monitorPerfect = runTest(topology, workloadPerfect) + +// assertAll( +// { assertEquals(600000, monitorNoDelay.maxTimestamp) { "The workload took longer to finish than expected." } }, +// { assertEquals(900000, monitorPerfect.maxTimestamp) { "The workload took longer to finish than expected." } }, +// +// { assertEquals(1000.0, monitorNoDelay.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, +// { assertEquals(3000.0, monitorNoDelay.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } }, +// { assertEquals(1000.0, monitorPerfect.taskCpuDemands["0"]?.get(0)) { "The cpu demanded by task 0 is incorrect" } }, +// { assertEquals(3000.0, monitorPerfect.taskCpuDemands["1"]?.get(0)) { "The cpu demanded by task 1 is incorrect" } }, +// +// { assertEquals(1000.0, monitorNoDelay.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, +// { assertEquals(3000.0, monitorNoDelay.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } }, +// { assertEquals(1000.0, monitorPerfect.taskCpuSupplied["0"]?.get(0)) { "The cpu supplied to task 0 is incorrect" } }, +// { assertEquals(3000.0, monitorPerfect.taskCpuSupplied["1"]?.get(0)) { "The cpu supplied to task 1 is incorrect" } }, +// ) + } +} diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt index 6d80ce56..f9a20c68 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/SchedulerTest.kt @@ -30,7 +30,7 @@ import org.opendc.compute.simulator.scheduler.filters.ComputeFilter import org.opendc.compute.simulator.scheduler.filters.RamFilter import org.opendc.compute.simulator.scheduler.filters.VCpuFilter import org.opendc.compute.workload.Task -import org.opendc.simulator.compute.workload.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceFragment import java.util.ArrayList class SchedulerTest { diff --git a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt index eadd97e4..df45f374 100644 --- a/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt +++ b/opendc-experiments/opendc-experiments-base/src/test/kotlin/org/opendc/experiments/base/TestingUtils.kt @@ -43,8 +43,10 @@ import org.opendc.compute.topology.specs.ClusterSpec import org.opendc.compute.workload.Task import org.opendc.experiments.base.experiment.specs.FailureModelSpec import org.opendc.experiments.base.runner.replay -import org.opendc.simulator.compute.workload.TraceFragment -import org.opendc.simulator.compute.workload.TraceWorkload +import org.opendc.simulator.compute.workload.trace.TraceFragment +import org.opendc.simulator.compute.workload.trace.TraceWorkload +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling +import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy import org.opendc.simulator.kotlin.runSimulation import java.time.Duration import java.time.LocalDateTime @@ -69,6 +71,7 @@ fun createTestTask( checkpointInterval: Long = 0L, checkpointDuration: Long = 0L, checkpointIntervalScaling: Double = 1.0, + scalingPolicy: ScalingPolicy = NoDelayScaling(), ): Task { return Task( UUID.nameUUIDFromBytes(name.toByteArray()), @@ -84,6 +87,7 @@ fun createTestTask( checkpointInterval, checkpointDuration, checkpointIntervalScaling, + scalingPolicy, ), ) } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkload.java index 2919fc3a..ebfcc552 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkload.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkload.java @@ -55,11 +55,11 @@ public abstract class SimWorkload extends FlowNode implements FlowConsumer { public abstract Workload getSnapshot(); - abstract void createCheckpointModel(); + public abstract void createCheckpointModel(); - abstract long getCheckpointInterval(); + public abstract long getCheckpointInterval(); - abstract long getCheckpointDuration(); + public abstract long getCheckpointDuration(); - abstract double getCheckpointIntervalScaling(); + public abstract double getCheckpointIntervalScaling(); } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimTraceWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/SimTraceWorkload.java index 9b12b1e3..93733268 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimTraceWorkload.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/SimTraceWorkload.java @@ -20,9 +20,12 @@ * SOFTWARE. */ -package org.opendc.simulator.compute.workload; +package org.opendc.simulator.compute.workload.trace; import java.util.LinkedList; +import org.opendc.simulator.compute.workload.SimWorkload; +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling; +import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy; import org.opendc.simulator.engine.graph.FlowConsumer; import org.opendc.simulator.engine.graph.FlowEdge; import org.opendc.simulator.engine.graph.FlowGraph; @@ -37,13 +40,18 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { private long startOfFragment; private FlowEdge machineEdge; - private double currentDemand; - private double currentSupply; + + private double cpuFreqDemand = 0.0; // The Cpu demanded by fragment + private double cpuFreqSupplied = 0.0; // The Cpu speed supplied + private double newCpuFreqSupplied = 0.0; // The Cpu speed supplied + private double remainingWork = 0.0; // The duration of the fragment at the demanded speed private long checkpointDuration; private TraceWorkload snapshot; + private ScalingPolicy scalingPolicy = new NoDelayScaling(); + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Basic Getters and Setters //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -57,27 +65,20 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { } @Override - long getCheckpointInterval() { + public long getCheckpointInterval() { return 0; } @Override - long getCheckpointDuration() { + public long getCheckpointDuration() { return 0; } @Override - double getCheckpointIntervalScaling() { + public double getCheckpointIntervalScaling() { return 0; } - public TraceFragment getNextFragment() { - this.currentFragment = this.remainingFragments.pop(); - this.fragmentIndex++; - - return this.currentFragment; - } - //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Constructors //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -87,15 +88,12 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { this.snapshot = workload; this.checkpointDuration = workload.getCheckpointDuration(); + this.scalingPolicy = workload.getScalingPolicy(); this.remainingFragments = new LinkedList<>(workload.getFragments()); this.fragmentIndex = 0; final FlowGraph graph = ((FlowNode) supplier).getGraph(); graph.addEdge(this, supplier); - - this.currentFragment = this.getNextFragment(); - pushOutgoingDemand(machineEdge, this.currentFragment.cpuUsage()); - this.startOfFragment = now; } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -105,36 +103,50 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { @Override public long onUpdate(long now) { long passedTime = getPassedTime(now); - long duration = this.currentFragment.duration(); + this.startOfFragment = now; - // The current Fragment has not yet been finished, continue - if (passedTime < duration) { - return now + (duration - passedTime); + // The amount of work done since last update + double finishedWork = this.scalingPolicy.getFinishedWork(this.cpuFreqDemand, this.cpuFreqSupplied, passedTime); + + this.remainingWork -= finishedWork; + + // If this.remainingWork <= 0, the fragment has been completed + if (this.remainingWork <= 0) { + this.startNextFragment(); + + this.invalidate(); + return Long.MAX_VALUE; } - // Loop through fragments until the passed time is filled. - // We need a while loop to account for skipping of fragments. - while (passedTime >= duration) { - if (this.remainingFragments.isEmpty()) { - this.stopWorkload(); - return Long.MAX_VALUE; - } + this.cpuFreqSupplied = this.newCpuFreqSupplied; - passedTime = passedTime - duration; + // The amount of time required to finish the fragment at this speed + long remainingDuration = this.scalingPolicy.getRemainingDuration( + this.cpuFreqDemand, this.newCpuFreqSupplied, this.remainingWork); - // get next Fragment - currentFragment = this.getNextFragment(); - duration = currentFragment.duration(); + return now + remainingDuration; + } + + public TraceFragment getNextFragment() { + if (this.remainingFragments.isEmpty()) { + return null; } + this.currentFragment = this.remainingFragments.pop(); + this.fragmentIndex++; - // start new fragment - this.startOfFragment = now - passedTime; + return this.currentFragment; + } - // Change the cpu Usage to the new Fragment - pushOutgoingDemand(machineEdge, this.currentFragment.cpuUsage()); + private void startNextFragment() { - // Return the time when the current fragment will complete - return this.startOfFragment + duration; + TraceFragment nextFragment = this.getNextFragment(); + if (nextFragment == null) { + this.stopWorkload(); + return; + } + double demand = nextFragment.cpuUsage(); + this.remainingWork = this.scalingPolicy.getRemainingWork(demand, nextFragment.duration()); + this.pushOutgoingDemand(this.machineEdge, demand); } @Override @@ -159,7 +171,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { * TODO: Maybe add checkpoint models for SimTraceWorkload */ @Override - void createCheckpointModel() {} + public void createCheckpointModel() {} /** * Create a new snapshot based on the current status of the workload. @@ -171,7 +183,15 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { // Get remaining time of current fragment long passedTime = getPassedTime(now); - long remainingTime = currentFragment.duration() - passedTime; + + // The amount of work done since last update + double finishedWork = this.scalingPolicy.getFinishedWork(this.cpuFreqDemand, this.cpuFreqSupplied, passedTime); + + this.remainingWork -= finishedWork; + + // The amount of time required to finish the fragment at this speed + long remainingTime = + this.scalingPolicy.getRemainingDuration(this.cpuFreqDemand, this.cpuFreqDemand, this.remainingWork); // If this is the end of the Task, don't make a snapshot if (remainingTime <= 0 && remainingFragments.isEmpty()) { @@ -189,13 +209,13 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { this.remainingFragments.addFirst(newFragment); // Create and add a fragment for processing the snapshot process - // TODO: improve the implementation of cpuUsage and coreCount - TraceFragment snapshotFragment = new TraceFragment(this.checkpointDuration, 123456, 1); + TraceFragment snapshotFragment = new TraceFragment( + this.checkpointDuration, this.snapshot.getMaxCpuDemand(), this.snapshot.getMaxCoreCount()); this.remainingFragments.addFirst(snapshotFragment); this.fragmentIndex = -1; - this.currentFragment = getNextFragment(); - pushOutgoingDemand(this.machineEdge, this.currentFragment.cpuUsage()); + startNextFragment(); + this.startOfFragment = now; this.invalidate(); @@ -213,11 +233,14 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { */ @Override public void handleIncomingSupply(FlowEdge supplierEdge, double newSupply) { - if (newSupply == this.currentSupply) { + if (newSupply == this.cpuFreqSupplied) { return; } - this.currentSupply = newSupply; + this.cpuFreqSupplied = this.newCpuFreqSupplied; + this.newCpuFreqSupplied = newSupply; + + this.invalidate(); } /** @@ -228,11 +251,11 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { */ @Override public void pushOutgoingDemand(FlowEdge supplierEdge, double newDemand) { - if (newDemand == this.currentDemand) { + if (newDemand == this.cpuFreqDemand) { return; } - this.currentDemand = newDemand; + this.cpuFreqDemand = newDemand; this.machineEdge.pushDemand(newDemand); } @@ -257,6 +280,7 @@ public class SimTraceWorkload extends SimWorkload implements FlowConsumer { if (this.machineEdge == null) { return; } + this.stopWorkload(); } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/TraceFragment.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceFragment.java index 550c2135..a09206a1 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/TraceFragment.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceFragment.java @@ -20,7 +20,7 @@ * SOFTWARE. */ -package org.opendc.simulator.compute.workload; +package org.opendc.simulator.compute.workload.trace; public record TraceFragment(long duration, double cpuUsage, int coreCount) { diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/TraceWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceWorkload.java index 7f82ab71..47292a7b 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/TraceWorkload.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/TraceWorkload.java @@ -20,10 +20,15 @@ * SOFTWARE. */ -package org.opendc.simulator.compute.workload; +package org.opendc.simulator.compute.workload.trace; import java.util.ArrayList; +import java.util.Comparator; import java.util.List; +import org.opendc.simulator.compute.workload.SimWorkload; +import org.opendc.simulator.compute.workload.Workload; +import org.opendc.simulator.compute.workload.trace.scaling.NoDelayScaling; +import org.opendc.simulator.compute.workload.trace.scaling.ScalingPolicy; import org.opendc.simulator.engine.graph.FlowSupplier; public class TraceWorkload implements Workload { @@ -31,20 +36,40 @@ public class TraceWorkload implements Workload { private final long checkpointInterval; private final long checkpointDuration; private final double checkpointIntervalScaling; + private final double maxCpuDemand; + private final int maxCoreCount; + + public ScalingPolicy getScalingPolicy() { + return scalingPolicy; + } + + private final ScalingPolicy scalingPolicy; public TraceWorkload( ArrayList<TraceFragment> fragments, long checkpointInterval, long checkpointDuration, - double checkpointIntervalScaling) { + double checkpointIntervalScaling, + ScalingPolicy scalingPolicy) { this.fragments = fragments; this.checkpointInterval = checkpointInterval; this.checkpointDuration = checkpointDuration; this.checkpointIntervalScaling = checkpointIntervalScaling; + this.scalingPolicy = scalingPolicy; + + // TODO: remove if we decide not to use it. + this.maxCpuDemand = fragments.stream() + .max(Comparator.comparing(TraceFragment::cpuUsage)) + .get() + .cpuUsage(); + this.maxCoreCount = fragments.stream() + .max(Comparator.comparing(TraceFragment::coreCount)) + .get() + .coreCount(); } public TraceWorkload(ArrayList<TraceFragment> fragments) { - this(fragments, 0L, 0L, 1.0); + this(fragments, 0L, 0L, 1.0, new NoDelayScaling()); } public ArrayList<TraceFragment> getFragments() { @@ -66,6 +91,14 @@ public class TraceWorkload implements Workload { return checkpointIntervalScaling; } + public int getMaxCoreCount() { + return maxCoreCount; + } + + public double getMaxCpuDemand() { + return maxCpuDemand; + } + public void removeFragments(int numberOfFragments) { if (numberOfFragments <= 0) { return; @@ -83,11 +116,15 @@ public class TraceWorkload implements Workload { } public static Builder builder() { - return builder(0L, 0L, 0.0); + return builder(0L, 0L, 0.0, new NoDelayScaling()); } - public static Builder builder(long checkpointInterval, long checkpointDuration, double checkpointIntervalScaling) { - return new Builder(checkpointInterval, checkpointDuration, checkpointIntervalScaling); + public static Builder builder( + long checkpointInterval, + long checkpointDuration, + double checkpointIntervalScaling, + ScalingPolicy scalingPolicy) { + return new Builder(checkpointInterval, checkpointDuration, checkpointIntervalScaling, scalingPolicy); } /** @@ -125,15 +162,21 @@ public class TraceWorkload implements Workload { private final long checkpointInterval; private final long checkpointDuration; private final double checkpointIntervalScaling; + private final ScalingPolicy scalingPolicy; /** * Construct a new {@link Builder} instance. */ - private Builder(long checkpointInterval, long checkpointDuration, double checkpointIntervalScaling) { + private Builder( + long checkpointInterval, + long checkpointDuration, + double checkpointIntervalScaling, + ScalingPolicy scalingPolicy) { this.fragments = new ArrayList<>(); this.checkpointInterval = checkpointInterval; this.checkpointDuration = checkpointDuration; this.checkpointIntervalScaling = checkpointIntervalScaling; + this.scalingPolicy = scalingPolicy; } /** @@ -152,7 +195,11 @@ public class TraceWorkload implements Workload { */ public TraceWorkload build() { return new TraceWorkload( - this.fragments, this.checkpointInterval, this.checkpointDuration, this.checkpointIntervalScaling); + this.fragments, + this.checkpointInterval, + this.checkpointDuration, + this.checkpointIntervalScaling, + this.scalingPolicy); } } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java new file mode 100644 index 00000000..4230bb55 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/NoDelayScaling.java @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute.workload.trace.scaling; + +/** + * The NoDelay scaling policy states that there will be no delay + * when less CPU can be provided than needed. + * + * This could be used in situations where the data is streamed. + * This will also result in the same behaviour as older OpenDC. + */ +public class NoDelayScaling implements ScalingPolicy { + @Override + public double getFinishedWork(double cpuFreqDemand, double cpuFreqSupplied, long passedTime) { + return cpuFreqDemand * passedTime; + } + + @Override + public long getRemainingDuration(double cpuFreqDemand, double cpuFreqSupplied, double remainingWork) { + return (long) (remainingWork / cpuFreqDemand); + } + + @Override + public double getRemainingWork(double cpuFreqDemand, long duration) { + return cpuFreqDemand * duration; + } +} diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java new file mode 100644 index 00000000..7eae70e6 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/PerfectScaling.java @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute.workload.trace.scaling; + +/** + * PerfectScaling scales the workload duration perfectly + * based on the CPU capacity. + * + * This means that if a fragment has a duration of 10 min at 4000 mHz, + * it will take 20 min and 2000 mHz. + */ +public class PerfectScaling implements ScalingPolicy { + @Override + public double getFinishedWork(double cpuFreqDemand, double cpuFreqSupplied, long passedTime) { + return cpuFreqSupplied * passedTime; + } + + @Override + public long getRemainingDuration(double cpuFreqDemand, double cpuFreqSupplied, double remainingWork) { + return (long) (remainingWork / cpuFreqSupplied); + } + + @Override + public double getRemainingWork(double cpuFreqDemand, long duration) { + return cpuFreqDemand * duration; + } +} diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java new file mode 100644 index 00000000..a0f473ba --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/trace/scaling/ScalingPolicy.java @@ -0,0 +1,59 @@ +/* + * Copyright (c) 2025 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute.workload.trace.scaling; + +/** + * Interface for the scaling policy. + * A scaling decides how a TaskFragment should scale when it is not getting the demanded capacity + */ +public interface ScalingPolicy { + + /** + * Calculate how much work was finished based on the demanded and supplied cpu + * + * @param cpuFreqDemand + * @param cpuFreqSupplied + * @param passedTime + * @return + */ + double getFinishedWork(double cpuFreqDemand, double cpuFreqSupplied, long passedTime); + + /** + * Calculate the remaining duration of this fragment based on the demanded and supplied cpu + * + * @param cpuFreqDemand + * @param cpuFreqSupplied + * @param remainingWork + * @return + */ + long getRemainingDuration(double cpuFreqDemand, double cpuFreqSupplied, double remainingWork); + + /** + * Calculate how much work is remaining based on the demanded and supplied cpu + * + * @param cpuFreqDemand + * @param duration + * @return + */ + double getRemainingWork(double cpuFreqDemand, long duration); +} diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/Coroutines.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/Coroutines.kt index ad69a3d6..49baaf48 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/Coroutines.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/Coroutines.kt @@ -24,7 +24,7 @@ package org.opendc.simulator.compute import kotlinx.coroutines.suspendCancellableCoroutine import org.opendc.simulator.compute.machine.SimMachine -import org.opendc.simulator.compute.workload.TraceWorkload +import org.opendc.simulator.compute.workload.trace.TraceWorkload import kotlin.coroutines.resume import kotlin.coroutines.resumeWithException diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/engine/FlowEngine.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/engine/FlowEngine.java index 67540f4e..3f18ac76 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/engine/FlowEngine.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/engine/FlowEngine.java @@ -139,16 +139,16 @@ public final class FlowEngine implements Runnable { * This method should only be invoked while inside an engine cycle. */ public void scheduleDelayedInContext(FlowNode ctx) { - FlowEventQueue timerQueue = this.eventQueue; - timerQueue.enqueue(ctx); + FlowEventQueue eventQueue = this.eventQueue; + eventQueue.enqueue(ctx); } /** * Run all the enqueued actions for the specified timestamp (<code>now</code>). */ private void doRunEngine(long now) { - final FlowCycleQueue queue = this.cycleQueue; - final FlowEventQueue timerQueue = this.eventQueue; + final FlowCycleQueue cycleQueue = this.cycleQueue; + final FlowEventQueue eventQueue = this.eventQueue; try { // Mark the engine as active to prevent concurrent calls to this method @@ -156,7 +156,7 @@ public final class FlowEngine implements Runnable { // Execute all scheduled updates at current timestamp while (true) { - final FlowNode ctx = timerQueue.poll(now); + final FlowNode ctx = eventQueue.poll(now); if (ctx == null) { break; } @@ -166,7 +166,7 @@ public final class FlowEngine implements Runnable { // Execute all immediate updates while (true) { - final FlowNode ctx = queue.poll(); + final FlowNode ctx = cycleQueue.poll(); if (ctx == null) { break; } @@ -178,7 +178,7 @@ public final class FlowEngine implements Runnable { } // Schedule an engine invocation for the next update to occur. - long headDeadline = timerQueue.peekDeadline(); + long headDeadline = eventQueue.peekDeadline(); if (headDeadline != Long.MAX_VALUE && headDeadline >= now) { trySchedule(futureInvocations, now, headDeadline); } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java index 16bb161f..ff7ff199 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowDistributor.java @@ -24,6 +24,9 @@ package org.opendc.simulator.engine.graph; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashSet; +import java.util.Objects; +import java.util.Set; public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsumer { private final ArrayList<FlowEdge> consumerEdges = new ArrayList<>(); @@ -36,13 +39,13 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu private double currentIncomingSupply; // The current supply provided by the supplier private boolean outgoingDemandUpdateNeeded = false; + private final Set<Integer> updatedDemands = + new HashSet<>(); // Array of consumers that updated their demand in this cycle private boolean overloaded = false; private double capacity; // What is the max capacity. Can probably be removed - private final ArrayList<Integer> updatedDemands = new ArrayList<>(); - public FlowDistributor(FlowGraph graph) { super(graph); } @@ -68,7 +71,9 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu return Long.MAX_VALUE; } - this.updateOutgoingSupplies(); + if (!this.outgoingSupplies.isEmpty()) { + this.updateOutgoingSupplies(); + } return Long.MAX_VALUE; } @@ -100,7 +105,7 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu // provide all consumers with their demand if (this.overloaded) { for (int idx = 0; idx < this.consumerEdges.size(); idx++) { - if (this.outgoingSupplies.get(idx) != this.incomingDemands.get(idx)) { + if (!Objects.equals(this.outgoingSupplies.get(idx), this.incomingDemands.get(idx))) { this.pushOutgoingSupply(this.consumerEdges.get(idx), this.incomingDemands.get(idx)); } } @@ -190,23 +195,25 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu this.totalIncomingDemand -= consumerEdge.getDemand(); + // Remove idx from consumers that updated their demands + this.updatedDemands.remove(idx); + this.consumerEdges.remove(idx); this.incomingDemands.remove(idx); this.outgoingSupplies.remove(idx); // update the consumer index for all consumerEdges higher than this. for (int i = idx; i < this.consumerEdges.size(); i++) { - this.consumerEdges.get(i).setConsumerIndex(i); + FlowEdge other = this.consumerEdges.get(i); + + other.setConsumerIndex(other.getConsumerIndex() - 1); } - for (int i = 0; i < this.updatedDemands.size(); i++) { - int j = this.updatedDemands.get(i); + for (int idx_other : this.updatedDemands) { - if (j == idx) { - this.updatedDemands.remove(idx); - } - if (j > idx) { - this.updatedDemands.set(i, j - 1); + if (idx_other > idx) { + this.updatedDemands.remove(idx_other); + this.updatedDemands.add(idx_other - 1); } } @@ -220,7 +227,9 @@ public class FlowDistributor extends FlowNode implements FlowSupplier, FlowConsu this.capacity = 0; this.currentIncomingSupply = 0; - this.invalidate(); + this.updatedDemands.clear(); + + this.closeNode(); } @Override diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowGraph.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowGraph.java index 0e6e137c..91662950 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowGraph.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/engine/graph/FlowGraph.java @@ -63,7 +63,7 @@ public class FlowGraph { // Remove all edges connected to node final ArrayList<FlowEdge> connectedEdges = nodeToEdge.get(node); - while (connectedEdges.size() > 0) { + while (!connectedEdges.isEmpty()) { removeEdge(connectedEdges.get(0)); } @@ -90,7 +90,7 @@ public class FlowGraph { throw new IllegalArgumentException("The consumer is not a node in this graph"); } if (!(this.nodes.contains((FlowNode) flowSupplier))) { - throw new IllegalArgumentException("The consumer is not a node in this graph"); + throw new IllegalArgumentException("The supplier is not a node in this graph"); } final FlowEdge flowEdge = new FlowEdge(flowConsumer, flowSupplier); |