diff options
Diffstat (limited to 'opendc-simulator/opendc-simulator-compute/src')
5 files changed, 220 insertions, 40 deletions
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimAbstractMachine.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimAbstractMachine.java index d968d884..40c219ed 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimAbstractMachine.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimAbstractMachine.java @@ -36,14 +36,11 @@ import org.opendc.simulator.flow2.Outlet; import org.opendc.simulator.flow2.sink.SimpleFlowSink; import org.opendc.simulator.flow2.util.FlowTransformer; import org.opendc.simulator.flow2.util.FlowTransforms; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Abstract implementation of the {@link SimMachine} interface. */ public abstract class SimAbstractMachine implements SimMachine { - private static final Logger LOGGER = LoggerFactory.getLogger(SimAbstractMachine.class); private final MachineModel model; private Context activeContext; @@ -108,7 +105,6 @@ public abstract class SimAbstractMachine implements SimMachine { private final Map<String, Object> meta; private final Consumer<Exception> completion; private boolean isClosed; - private Exception cause; /** * Construct a new {@link Context} instance. @@ -158,6 +154,11 @@ public abstract class SimAbstractMachine implements SimMachine { @Override public final void shutdown() { + shutdown(null); + } + + @Override + public final void shutdown(Exception cause) { if (isClosed) { return; } @@ -170,19 +171,17 @@ public abstract class SimAbstractMachine implements SimMachine { // Cancel all the resources associated with the machine doCancel(); - Exception e = this.cause; - try { workload.onStop(this); - } catch (Exception cause) { - if (e != null) { - e.addSuppressed(cause); + } catch (Exception e) { + if (cause == null) { + cause = e; } else { - e = cause; + cause.addSuppressed(e); } } - completion.accept(e); + completion.accept(cause); } /** @@ -193,8 +192,7 @@ public abstract class SimAbstractMachine implements SimMachine { machine.activeContext = this; workload.onStart(this); } catch (Exception cause) { - this.cause = cause; - shutdown(); + shutdown(cause); } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimMachineContext.java b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimMachineContext.java index 5d08e2b7..0f7d4c8b 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimMachineContext.java +++ b/opendc-simulator/opendc-simulator-compute/src/main/java/org/opendc/simulator/compute/SimMachineContext.java @@ -24,6 +24,7 @@ package org.opendc.simulator.compute; import java.util.List; import java.util.Map; +import java.util.function.Consumer; import org.opendc.simulator.compute.workload.SimWorkload; import org.opendc.simulator.flow2.FlowGraph; @@ -43,7 +44,7 @@ public interface SimMachineContext { /** * Return the metadata associated with the context. * <p> - * Users can pass this metadata to the workload via {@link SimMachine#startWorkload(SimWorkload, Map)}. + * Users can pass this metadata to the workload via {@link SimMachine#startWorkload(SimWorkload, Map, Consumer)}. */ Map<String, Object> getMeta(); @@ -76,4 +77,11 @@ public interface SimMachineContext { * Shutdown the workload. */ void shutdown(); + + /** + * Shutdown the workload due to failure. + * + * @param cause The cause for shutting down the workload. + */ + void shutdown(Exception cause); } 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 index 9304122a..f838328b 100644 --- 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 @@ -30,15 +30,11 @@ 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; @@ -65,9 +61,7 @@ final class SimChainWorkload implements SimWorkload { final Context context = new Context(ctx); activeContext = context; - if (!context.doStart(workloads[activeWorkloadIndex])) { - ctx.shutdown(); - } + tryThrow(context.doStart(workloads[activeWorkloadIndex])); } @Override @@ -82,7 +76,7 @@ final class SimChainWorkload implements SimWorkload { final Context context = activeContext; activeContext = null; - context.doStop(workloads[activeWorkloadIndex]); + tryThrow(context.doStop(workloads[activeWorkloadIndex])); } /** @@ -132,51 +126,80 @@ final class SimChainWorkload implements SimWorkload { @Override public void shutdown() { + shutdown(null); + } + + @Override + public void shutdown(Exception cause) { final SimWorkload[] workloads = SimChainWorkload.this.workloads; final int activeWorkloadIndex = ++SimChainWorkload.this.activeWorkloadIndex; - if (doStop(workloads[activeWorkloadIndex - 1]) && activeWorkloadIndex < workloads.length) { + final Exception stopException = doStop(workloads[activeWorkloadIndex - 1]); + if (cause == null) { + cause = stopException; + } else if (stopException != null) { + cause.addSuppressed(stopException); + } + + if (stopException == null && activeWorkloadIndex < workloads.length) { ctx.reset(); - if (doStart(workloads[activeWorkloadIndex])) { + final Exception startException = doStart(workloads[activeWorkloadIndex]); + + if (startException == null) { return; + } else if (cause == null) { + cause = startException; + } else { + cause.addSuppressed(startException); } } - ctx.shutdown(); + ctx.shutdown(cause); } /** * Start the specified workload. * - * @return <code>true</code> if the workload started successfully, <code>false</code> otherwise. + * @return The {@link Exception} that occurred while starting the workload or <code>null</code> if the workload + * started successfully. */ - private boolean doStart(SimWorkload workload) { + private Exception doStart(SimWorkload workload) { try { workload.onStart(this); } catch (Exception cause) { - LOGGER.warn("Workload failed during onStart callback", cause); - doStop(workload); - return false; + final Exception stopException = doStop(workload); + if (stopException != null) { + cause.addSuppressed(stopException); + } + return cause; } - return true; + return null; } /** * Stop the specified workload. * - * @return <code>true</code> if the workload stopped successfully, <code>false</code> otherwise. + * @return The {@link Exception} that occurred while stopping the workload or <code>null</code> if the workload + * stopped successfully. */ - private boolean doStop(SimWorkload workload) { + private Exception doStop(SimWorkload workload) { try { workload.onStop(this); } catch (Exception cause) { - LOGGER.warn("Workload failed during onStop callback", cause); - return false; + return cause; } - return true; + return null; + } + } + + @SuppressWarnings("unchecked") + private static <T extends Throwable> void tryThrow(Throwable e) throws T { + if (e == null) { + return; } + throw (T) e; } } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt index 266839bd..cde6763c 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt @@ -22,6 +22,8 @@ package org.opendc.simulator.compute +import io.mockk.every +import io.mockk.mockk import kotlinx.coroutines.CancellationException import kotlinx.coroutines.cancel import kotlinx.coroutines.coroutineScope @@ -334,4 +336,71 @@ class SimMachineTest { } } } + + @Test + fun testCatchStartFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workload = mockk<SimWorkload>() + every { workload.onStart(any()) } throws IllegalStateException() + + assertThrows<IllegalStateException> { machine.runWorkload(workload) } + } + + @Test + fun testCatchStopFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workload = mockk<SimWorkload>() + every { workload.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown() } + every { workload.onStop(any()) } throws IllegalStateException() + + assertThrows<IllegalStateException> { machine.runWorkload(workload) } + } + + @Test + fun testCatchShutdownFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workload = mockk<SimWorkload>() + every { workload.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) } + + assertThrows<IllegalStateException> { machine.runWorkload(workload) } + } + + @Test + fun testCatchNestedFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workload = mockk<SimWorkload>() + every { workload.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) } + every { workload.onStop(any()) } throws IllegalStateException() + + val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) } + assertEquals(1, exc.cause!!.suppressedExceptions.size) + } } 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 index 6bf05f65..2210374d 100644 --- 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 @@ -28,7 +28,9 @@ 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.junit.jupiter.api.assertThrows import org.opendc.simulator.compute.SimBareMetalMachine +import org.opendc.simulator.compute.SimMachineContext import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.compute.model.MemoryUnit import org.opendc.simulator.compute.model.ProcessingNode @@ -94,7 +96,7 @@ class SimChainWorkloadTest { SimRuntimeWorkload(1000, 1.0) ) - machine.runWorkload(workload) + assertThrows<IllegalStateException> { machine.runWorkload(workload) } assertEquals(0, clock.millis()) } @@ -120,7 +122,7 @@ class SimChainWorkloadTest { SimRuntimeWorkload(1000, 1.0) ) - machine.runWorkload(workload) + assertThrows<IllegalStateException> { machine.runWorkload(workload) } assertEquals(1000, clock.millis()) } @@ -144,7 +146,7 @@ class SimChainWorkloadTest { SimRuntimeWorkload(1000, 1.0) ) - machine.runWorkload(workload) + assertThrows<IllegalStateException> { machine.runWorkload(workload) } assertEquals(1000, clock.millis()) } @@ -169,8 +171,88 @@ class SimChainWorkloadTest { SimRuntimeWorkload(1000, 1.0) ) - machine.runWorkload(workload) + assertThrows<IllegalStateException> { machine.runWorkload(workload) } assertEquals(2000, clock.millis()) } + + @Test + fun testStartAndStopFailure() = 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() + every { workloadA.onStop(any()) } throws IllegalStateException() + + val workload = + SimWorkloads.chain( + SimRuntimeWorkload(1000, 1.0), + workloadA + ) + + val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) } + + assertEquals(2, exc.cause!!.suppressedExceptions.size) + assertEquals(1000, clock.millis()) + } + + @Test + fun testShutdownAndStopFailure() = 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()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) } + every { workloadA.onStop(any()) } throws IllegalStateException() + + val workload = + SimWorkloads.chain( + SimRuntimeWorkload(1000, 1.0), + workloadA + ) + + val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) } + + assertEquals(1, exc.cause!!.suppressedExceptions.size) + assertEquals(1000, clock.millis()) + } + + @Test + fun testShutdownAndStartFailure() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val machine = SimBareMetalMachine.create( + graph, + machineModel + ) + + val workloadA = mockk<SimWorkload>(relaxUnitFun = true) + every { workloadA.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) } + + val workloadB = mockk<SimWorkload>(relaxUnitFun = true) + every { workloadB.onStart(any()) } throws IllegalStateException() + + val workload = + SimWorkloads.chain( + SimRuntimeWorkload(1000, 1.0), + workloadA, + workloadB + ) + + val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) } + assertEquals(1, exc.cause!!.suppressedExceptions.size) + assertEquals(1000, clock.millis()) + } } |