diff options
3 files changed, 365 insertions, 0 deletions
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java new file mode 100644 index 00000000..9304122a --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimChainWorkload.java @@ -0,0 +1,182 @@ +/* + * Copyright (c) 2022 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; + +import java.util.List; +import java.util.Map; +import org.opendc.simulator.compute.SimMachineContext; +import org.opendc.simulator.compute.SimMemory; +import org.opendc.simulator.compute.SimNetworkInterface; +import org.opendc.simulator.compute.SimProcessingUnit; +import org.opendc.simulator.compute.SimStorageInterface; +import org.opendc.simulator.flow2.FlowGraph; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link SimWorkload} that composes two {@link SimWorkload}s. + */ +final class SimChainWorkload implements SimWorkload { + private static final Logger LOGGER = LoggerFactory.getLogger(SimChainWorkload.class); + + private final SimWorkload[] workloads; + private int activeWorkloadIndex; + + private Context activeContext; + + /** + * Construct a {@link SimChainWorkload} instance. + * + * @param workloads The workloads to chain. + */ + SimChainWorkload(SimWorkload... workloads) { + this.workloads = workloads; + } + + @Override + public void onStart(SimMachineContext ctx) { + final SimWorkload[] workloads = this.workloads; + final int activeWorkloadIndex = this.activeWorkloadIndex; + + if (activeWorkloadIndex >= workloads.length) { + return; + } + + final Context context = new Context(ctx); + activeContext = context; + + if (!context.doStart(workloads[activeWorkloadIndex])) { + ctx.shutdown(); + } + } + + @Override + public void onStop(SimMachineContext ctx) { + final SimWorkload[] workloads = this.workloads; + final int activeWorkloadIndex = this.activeWorkloadIndex; + + if (activeWorkloadIndex >= workloads.length) { + return; + } + + final Context context = activeContext; + activeContext = null; + + context.doStop(workloads[activeWorkloadIndex]); + } + + /** + * A {@link SimMachineContext} that intercepts the shutdown calls. + */ + private class Context implements SimMachineContext { + private final SimMachineContext ctx; + + private Context(SimMachineContext ctx) { + this.ctx = ctx; + } + + @Override + public FlowGraph getGraph() { + return ctx.getGraph(); + } + + @Override + public Map<String, Object> getMeta() { + return ctx.getMeta(); + } + + @Override + public List<? extends SimProcessingUnit> getCpus() { + return ctx.getCpus(); + } + + @Override + public SimMemory getMemory() { + return ctx.getMemory(); + } + + @Override + public List<? extends SimNetworkInterface> getNetworkInterfaces() { + return ctx.getNetworkInterfaces(); + } + + @Override + public List<? extends SimStorageInterface> getStorageInterfaces() { + return ctx.getStorageInterfaces(); + } + + @Override + public void reset() { + ctx.reset(); + } + + @Override + public void shutdown() { + final SimWorkload[] workloads = SimChainWorkload.this.workloads; + final int activeWorkloadIndex = ++SimChainWorkload.this.activeWorkloadIndex; + + if (doStop(workloads[activeWorkloadIndex - 1]) && activeWorkloadIndex < workloads.length) { + ctx.reset(); + + if (doStart(workloads[activeWorkloadIndex])) { + return; + } + } + + ctx.shutdown(); + } + + /** + * Start the specified workload. + * + * @return <code>true</code> if the workload started successfully, <code>false</code> otherwise. + */ + private boolean doStart(SimWorkload workload) { + try { + workload.onStart(this); + } catch (Exception cause) { + LOGGER.warn("Workload failed during onStart callback", cause); + doStop(workload); + return false; + } + + return true; + } + + /** + * Stop the specified workload. + * + * @return <code>true</code> if the workload stopped successfully, <code>false</code> otherwise. + */ + private boolean doStop(SimWorkload workload) { + try { + workload.onStop(this); + } catch (Exception cause) { + LOGGER.warn("Workload failed during onStop callback", cause); + return false; + } + + return true; + } + } +} diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkloads.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkloads.java index 8655fd0a..82557d06 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkloads.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/workload/SimWorkloads.java @@ -60,4 +60,11 @@ public class SimWorkloads { public static SimWorkload runtime(Duration duration, double utilization) { return runtime(duration.toMillis(), utilization); } + + /** + * Chain the specified <code>workloads</code> into a single {@link SimWorkload}. + */ + public static SimWorkload chain(SimWorkload... workloads) { + return new SimChainWorkload(workloads); + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt new file mode 100644 index 00000000..6bf05f65 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt @@ -0,0 +1,176 @@ +/* + * Copyright (c) 2022 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 + +import io.mockk.every +import io.mockk.mockk +import io.mockk.spyk +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test +import org.opendc.simulator.compute.SimBareMetalMachine +import org.opendc.simulator.compute.model.MachineModel +import org.opendc.simulator.compute.model.MemoryUnit +import org.opendc.simulator.compute.model.ProcessingNode +import org.opendc.simulator.compute.model.ProcessingUnit +import org.opendc.simulator.compute.runWorkload +import org.opendc.simulator.flow2.FlowEngine +import org.opendc.simulator.kotlin.runSimulation + +/** + * Test suite for the [SimChainWorkload] class. + */ +class SimChainWorkloadTest { + private lateinit var machineModel: MachineModel + + @BeforeEach + fun setUp() { + val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2) + + machineModel = MachineModel( + /*cpus*/ List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 1000.0) }, + /*memory*/ List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) } + ) + } + + @Test + fun testMultipleWorkloads() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workload = + SimWorkloads.chain( + SimRuntimeWorkload(1000, 1.0), + SimRuntimeWorkload(1000, 1.0) + ) + + machine.runWorkload(workload) + + assertEquals(2000, clock.millis()) + } + + @Test + fun testStartFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workloadA = mockk<SimWorkload>() + every { workloadA.onStart(any()) } throws IllegalStateException("Staged") + every { workloadA.onStop(any()) } returns Unit + + val workload = + SimWorkloads.chain( + workloadA, + SimRuntimeWorkload(1000, 1.0) + ) + + machine.runWorkload(workload) + + assertEquals(0, clock.millis()) + } + + @Test + fun testStartFailureSecond() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workloadA = mockk<SimWorkload>() + every { workloadA.onStart(any()) } throws IllegalStateException("Staged") + every { workloadA.onStop(any()) } returns Unit + + val workload = + SimWorkloads.chain( + SimRuntimeWorkload(1000, 1.0), + workloadA, + SimRuntimeWorkload(1000, 1.0) + ) + + machine.runWorkload(workload) + + assertEquals(1000, clock.millis()) + } + + @Test + fun testStopFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workloadA = spyk<SimWorkload>(SimRuntimeWorkload(1000, 1.0)) + every { workloadA.onStop(any()) } throws IllegalStateException("Staged") + + val workload = + SimWorkloads.chain( + workloadA, + SimRuntimeWorkload(1000, 1.0) + ) + + machine.runWorkload(workload) + + assertEquals(1000, clock.millis()) + } + + @Test + fun testStopFailureSecond() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workloadA = spyk<SimWorkload>(SimRuntimeWorkload(1000, 1.0)) + every { workloadA.onStop(any()) } throws IllegalStateException("Staged") + + val workload = + SimWorkloads.chain( + SimRuntimeWorkload(1000, 1.0), + workloadA, + SimRuntimeWorkload(1000, 1.0) + ) + + machine.runWorkload(workload) + + assertEquals(2000, clock.millis()) + } +} |