diff options
Diffstat (limited to 'opendc-simulator/opendc-simulator-compute')
16 files changed, 431 insertions, 314 deletions
diff --git a/opendc-simulator/opendc-simulator-compute/build.gradle.kts b/opendc-simulator/opendc-simulator-compute/build.gradle.kts index a2bb89c2..ca8b912a 100644 --- a/opendc-simulator/opendc-simulator-compute/build.gradle.kts +++ b/opendc-simulator/opendc-simulator-compute/build.gradle.kts @@ -35,7 +35,7 @@ dependencies { api(projects.opendcSimulator.opendcSimulatorPower) api(projects.opendcSimulator.opendcSimulatorNetwork) implementation(projects.opendcSimulator.opendcSimulatorCore) - implementation(projects.opendcUtils) + implementation(libs.kotlin.logging) testImplementation(libs.slf4j.simple) } diff --git a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt index cb52d24f..91e91f9d 100644 --- a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt +++ b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt @@ -76,7 +76,7 @@ class SimMachineBenchmarks { val machine = SimBareMetalMachine( engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) - return@runBlockingSimulation machine.run(SimTraceWorkload(trace)) + return@runBlockingSimulation machine.runWorkload(SimTraceWorkload(trace)) } } @@ -89,15 +89,15 @@ class SimMachineBenchmarks { ) val hypervisor = SimSpaceSharedHypervisor(engine, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } - val vm = hypervisor.createMachine(machineModel) + val vm = hypervisor.newMachine(machineModel) try { - return@runBlockingSimulation vm.run(SimTraceWorkload(trace)) + return@runBlockingSimulation vm.runWorkload(SimTraceWorkload(trace)) } finally { - vm.close() - machine.close() + vm.cancel() + machine.cancel() } } } @@ -111,15 +111,15 @@ class SimMachineBenchmarks { ) val hypervisor = SimFairShareHypervisor(engine, null, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } - val vm = hypervisor.createMachine(machineModel) + val vm = hypervisor.newMachine(machineModel) try { - return@runBlockingSimulation vm.run(SimTraceWorkload(trace)) + return@runBlockingSimulation vm.runWorkload(SimTraceWorkload(trace)) } finally { - vm.close() - machine.close() + vm.cancel() + machine.cancel() } } } @@ -133,22 +133,22 @@ class SimMachineBenchmarks { ) val hypervisor = SimFairShareHypervisor(engine, null, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } coroutineScope { repeat(2) { - val vm = hypervisor.createMachine(machineModel) + val vm = hypervisor.newMachine(machineModel) launch { try { - vm.run(SimTraceWorkload(trace)) + vm.runWorkload(SimTraceWorkload(trace)) } finally { - machine.close() + machine.cancel() } } } } - machine.close() + machine.cancel() } } } 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 new file mode 100644 index 00000000..c23f48dc --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/Coroutines.kt @@ -0,0 +1,69 @@ +/* + * Copyright (c) 2021 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 + +import kotlinx.coroutines.suspendCancellableCoroutine +import org.opendc.simulator.compute.workload.SimWorkload +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException + +/** + * Run the specified [SimWorkload] on this machine and suspend execution util [workload] has finished. + * + * @param workload The workload to start on the machine. + * @param meta The metadata to pass to the workload. + * @return A [SimMachineContext] that represents the execution context for the workload. + * @throws IllegalStateException if a workload is already active on the machine or if the machine is closed. + */ +public suspend fun SimMachine.runWorkload(workload: SimWorkload, meta: Map<String, Any> = emptyMap()) { + return suspendCancellableCoroutine { cont -> + cont.invokeOnCancellation { this@runWorkload.cancel() } + + startWorkload( + object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + try { + workload.onStart(ctx) + } catch (cause: Throwable) { + cont.resumeWithException(cause) + throw cause + } + } + + override fun onStop(ctx: SimMachineContext) { + try { + workload.onStop(ctx) + + if (!cont.isCompleted) { + cont.resume(Unit) + } + } catch (cause: Throwable) { + cont.resumeWithException(cause) + throw cause + } + } + }, + meta + ) + } +} diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt index 5909d980..6a4c594d 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt @@ -23,6 +23,7 @@ package org.opendc.simulator.compute import kotlinx.coroutines.* +import mu.KotlinLogging import org.opendc.simulator.compute.device.SimNetworkAdapter import org.opendc.simulator.compute.device.SimPeripheral import org.opendc.simulator.compute.model.MachineModel @@ -31,8 +32,6 @@ import org.opendc.simulator.compute.model.NetworkAdapter import org.opendc.simulator.compute.model.StorageDevice import org.opendc.simulator.compute.workload.SimWorkload import org.opendc.simulator.flow.* -import kotlin.coroutines.Continuation -import kotlin.coroutines.resume /** * Abstract implementation of the [SimMachine] interface. @@ -72,55 +71,19 @@ public abstract class SimAbstractMachine( public override val peripherals: List<SimPeripheral> = net.map { it as SimNetworkAdapter } /** - * A flag to indicate that the machine is terminated. - */ - private var isTerminated = false - - /** * The current active [Context]. */ private var _ctx: Context? = null - /** - * This method is invoked when the machine is started. - */ - protected open fun onStart(ctx: SimMachineContext) {} - - /** - * This method is invoked when the machine is stopped. - */ - protected open fun onStop(ctx: SimMachineContext) { - _ctx = null - } - - /** - * Converge the specified [SimWorkload] on this machine and suspend execution util the workload has finished. - */ - override suspend fun run(workload: SimWorkload, meta: Map<String, Any>) { - check(!isTerminated) { "Machine is terminated" } + override fun startWorkload(workload: SimWorkload, meta: Map<String, Any>): SimMachineContext { check(_ctx == null) { "A machine cannot run concurrently" } - return suspendCancellableCoroutine { cont -> - val ctx = Context(meta, cont) - _ctx = ctx - - // Cancel all cpus on cancellation - cont.invokeOnCancellation { ctx.close() } - - engine.batch { - onStart(ctx) - - workload.onStart(ctx) - } - } + val ctx = Context(workload, meta) + ctx.start() + return ctx } - override fun close() { - if (isTerminated) { - return - } - - isTerminated = true + override fun cancel() { _ctx?.close() } @@ -130,15 +93,33 @@ public abstract class SimAbstractMachine( /** * The execution context in which the workload runs. + * + * @param workload The workload that is running on the machine. + * @param meta The metadata passed to the workload. */ - private inner class Context(override val meta: Map<String, Any>, private val cont: Continuation<Unit>) : SimMachineContext { + private inner class Context( + private val workload: SimWorkload, + override val meta: Map<String, Any> + ) : SimMachineContext { /** * A flag to indicate that the context has been closed. */ private var isClosed = false - override val engine: FlowEngine - get() = this@SimAbstractMachine.engine + override val engine: FlowEngine = this@SimAbstractMachine.engine + + /** + * Start this context. + */ + fun start() { + try { + _ctx = this + engine.batch { workload.onStart(this) } + } catch (cause: Throwable) { + logger.warn(cause) { "Workload failed during onStart callback" } + close() + } + } override val cpus: List<SimProcessingUnit> = this@SimAbstractMachine.cpus @@ -154,15 +135,43 @@ public abstract class SimAbstractMachine( } isClosed = true + assert(_ctx == this) { "Invariant violation: multiple contexts active for a single machine" } + _ctx = null + + // Cancel all the resources associated with the machine + doCancel() + + try { + workload.onStop(this) + } catch (cause: Throwable) { + logger.warn(cause) { "Workload failed during onStop callback" } + } + } + + /** + * Run the stop procedures for the resources associated with the machine. + */ + private fun doCancel() { engine.batch { for (cpu in cpus) { cpu.cancel() } - } - onStop(this) - cont.resume(Unit) + memory.cancel() + + for (ifx in net) { + (ifx as NetworkAdapterImpl).disconnect() + } + + for (storage in storage) { + val impl = storage as StorageDeviceImpl + impl.read.cancel() + impl.write.cancel() + } + } } + + override fun toString(): String = "SimAbstractMachine.Context" } /** @@ -218,4 +227,12 @@ public abstract class SimAbstractMachine( override fun toString(): String = "SimAbstractMachine.StorageDeviceImpl[name=$name,capacity=$capacity]" } + + private companion object { + /** + * The logging instance associated with this class. + */ + @JvmStatic + private val logger = KotlinLogging.logger {} + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt index ab0b56ae..94581e89 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt @@ -29,7 +29,7 @@ import org.opendc.simulator.compute.workload.SimWorkload /** * A generic machine that is able to run a [SimWorkload]. */ -public interface SimMachine : AutoCloseable { +public interface SimMachine { /** * The model of the machine containing its specifications. */ @@ -41,12 +41,19 @@ public interface SimMachine : AutoCloseable { public val peripherals: List<SimPeripheral> /** - * Converge the specified [SimWorkload] on this machine and suspend execution util the workload has finished. + * Start the specified [SimWorkload] on this machine. + * + * @param workload The workload to start on the machine. + * @param meta The metadata to pass to the workload. + * @return A [SimMachineContext] that represents the execution context for the workload. + * @throws IllegalStateException if a workload is already active on the machine or if the machine is closed. */ - public suspend fun run(workload: SimWorkload, meta: Map<String, Any> = emptyMap()) + public fun startWorkload(workload: SimWorkload, meta: Map<String, Any> = emptyMap()): SimMachineContext /** - * Terminate this machine. + * Cancel the workload that is currently running on this machine. + * + * If no workload is active, this operation is a no-op. */ - public override fun close() + public fun cancel() } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt index eda59d2d..07465126 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt @@ -28,6 +28,7 @@ import org.opendc.simulator.compute.kernel.cpufreq.ScalingPolicy import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.compute.model.ProcessingUnit +import org.opendc.simulator.compute.workload.SimWorkload import org.opendc.simulator.flow.* import org.opendc.simulator.flow.interference.InterferenceKey import org.opendc.simulator.flow.mux.FlowMultiplexer @@ -93,13 +94,20 @@ public abstract class SimAbstractHypervisor( private val governors = mutableListOf<ScalingGovernor.Logic>() /* SimHypervisor */ - override fun createMachine(model: MachineModel, interferenceId: String?): SimVirtualMachine { + override fun newMachine(model: MachineModel, interferenceId: String?): SimVirtualMachine { require(canFit(model)) { "Machine does not fit" } val vm = VirtualMachine(model, interferenceId) _vms.add(vm) return vm } + override fun removeMachine(machine: SimVirtualMachine) { + if (_vms.remove(machine)) { + // This cast must always succeed, since `_vms` only contains `VirtualMachine` types. + (machine as VirtualMachine).close() + } + } + /* SimWorkload */ override fun onStart(ctx: SimMachineContext) { context = ctx @@ -122,6 +130,8 @@ public abstract class SimAbstractHypervisor( } } + override fun onStop(ctx: SimMachineContext) {} + private var _cpuCount = 0 private var _cpuCapacity = 0.0 @@ -145,11 +155,16 @@ public abstract class SimAbstractHypervisor( private inner class VirtualMachine( model: MachineModel, interferenceId: String? = null - ) : SimAbstractMachine(engine, parent = null, model), SimVirtualMachine { + ) : SimAbstractMachine(engine, parent = null, model), SimVirtualMachine, AutoCloseable { + /** + * A flag to indicate that the machine is closed. + */ + private var isClosed = false + /** * The interference key of this virtual machine. */ - private var interferenceKey: InterferenceKey? = interferenceId?.let { interferenceDomain?.createKey(it) } + private val interferenceKey: InterferenceKey? = interferenceId?.let { interferenceDomain?.createKey(it) } /** * The vCPUs of the machine. @@ -181,36 +196,60 @@ public abstract class SimAbstractHypervisor( override val cpuUsage: Double get() = cpus.sumOf(FlowConsumer::rate) - override fun onStart(ctx: SimMachineContext) { - val interferenceKey = interferenceKey - if (interferenceKey != null) { - interferenceDomain?.join(interferenceKey) - } - - super.onStart(ctx) + override fun startWorkload(workload: SimWorkload, meta: Map<String, Any>): SimMachineContext { + check(!isClosed) { "Machine is closed" } + + return super.startWorkload( + object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + try { + joinInterferenceDomain() + workload.onStart(ctx) + } catch (cause: Throwable) { + leaveInterferenceDomain() + throw cause + } + } + + override fun onStop(ctx: SimMachineContext) { + leaveInterferenceDomain() + workload.onStop(ctx) + } + }, + meta + ) } - override fun onStop(ctx: SimMachineContext) { - super.onStop(ctx) - - val interferenceKey = interferenceKey - if (interferenceKey != null) { - interferenceDomain?.leave(interferenceKey) + override fun close() { + if (isClosed) { + return } - } - override fun close() { - super.close() + isClosed = true + cancel() for (cpu in cpus) { cpu.close() } + } - _vms.remove(this) + /** + * Join the interference domain of the hypervisor. + */ + private fun joinInterferenceDomain() { + val interferenceKey = interferenceKey + if (interferenceKey != null) { + interferenceDomain?.join(interferenceKey) + } + } + /** + * Leave the interference domain of the hypervisor. + */ + private fun leaveInterferenceDomain() { val interferenceKey = interferenceKey if (interferenceKey != null) { - interferenceDomain?.removeKey(interferenceKey) + interferenceDomain?.leave(interferenceKey) } } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt index 57d4cf20..a69f419f 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt @@ -67,5 +67,12 @@ public interface SimHypervisor : SimWorkload { * @param model The machine to create. * @param interferenceId An identifier for the interference model. */ - public fun createMachine(model: MachineModel, interferenceId: String? = null): SimVirtualMachine + public fun newMachine(model: MachineModel, interferenceId: String? = null): SimVirtualMachine + + /** + * Remove the specified [machine] from the hypervisor. + * + * @param machine The machine to remove. + */ + public fun removeMachine(machine: SimVirtualMachine) } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt index 99f4a1e1..726d1f56 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt @@ -48,5 +48,7 @@ public class SimFlopsWorkload( } } + override fun onStop(ctx: SimMachineContext) {} + override fun toString(): String = "SimFlopsWorkload(FLOPs=$flops,utilization=$utilization)" } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt index 2ef3bc43..8a3f5f84 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt @@ -48,5 +48,7 @@ public class SimRuntimeWorkload( } } + override fun onStop(ctx: SimMachineContext) {} + override fun toString(): String = "SimRuntimeWorkload(duration=$duration,utilization=$utilization)" } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt index 53c98409..ce04a790 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt @@ -40,5 +40,7 @@ public class SimTraceWorkload(private val trace: SimTrace, private val offset: L } } + override fun onStop(ctx: SimMachineContext) {} + override fun toString(): String = "SimTraceWorkload" } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt index b80665fa..61c6e2ad 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt @@ -37,4 +37,11 @@ public interface SimWorkload { * @param ctx The execution context in which the machine runs. */ public fun onStart(ctx: SimMachineContext) + + /** + * This method is invoked when the workload is stopped. + * + * @param ctx The execution context in which the machine runs. + */ + public fun onStop(ctx: SimMachineContext) } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt index cc4f1f6a..742470a1 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt @@ -33,31 +33,50 @@ public class SimWorkloadLifecycle(private val ctx: SimMachineContext) { /** * The resource consumers which represent the lifecycle of the workload. */ - private val waiting = mutableSetOf<FlowSource>() + private val waiting = HashSet<Wrapper>() /** - * Wait for the specified [consumer] to complete before ending the lifecycle of the workload. + * Wait for the specified [source] to complete before ending the lifecycle of the workload. */ - public fun waitFor(consumer: FlowSource): FlowSource { - waiting.add(consumer) - return object : FlowSource by consumer { - override fun onStop(conn: FlowConnection, now: Long, delta: Long) { - try { - consumer.onStop(conn, now, delta) - } finally { - complete(consumer) - } - } - override fun toString(): String = "SimWorkloadLifecycle.Consumer[delegate=$consumer]" - } + public fun waitFor(source: FlowSource): FlowSource { + val wrapper = Wrapper(source) + waiting.add(wrapper) + return wrapper } /** - * Complete the specified [FlowSource]. + * Complete the specified [Wrapper]. */ - private fun complete(consumer: FlowSource) { - if (waiting.remove(consumer) && waiting.isEmpty()) { + private fun complete(wrapper: Wrapper) { + if (waiting.remove(wrapper) && waiting.isEmpty()) { ctx.close() } } + + /** + * A [FlowSource] that wraps [delegate] and informs [SimWorkloadLifecycle] that is has completed. + */ + private inner class Wrapper(private val delegate: FlowSource) : FlowSource { + override fun onStart(conn: FlowConnection, now: Long) { + delegate.onStart(conn, now) + } + + override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long { + return delegate.onPull(conn, now, delta) + } + + override fun onConverge(conn: FlowConnection, now: Long, delta: Long) { + delegate.onConverge(conn, now, delta) + } + + override fun onStop(conn: FlowConnection, now: Long, delta: Long) { + try { + delegate.onStop(conn, now, delta) + } finally { + complete(this) + } + } + + override fun toString(): String = "SimWorkloadLifecycle.Wrapper[delegate=$delegate]" + } } 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 0bb24ed8..644eb497 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 @@ -65,14 +65,10 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) + machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) - // Two cores execute 1000 MFlOps per second (1000 ms) - assertEquals(1000, clock.millis()) - } finally { - machine.close() - } + // Two cores execute 1000 MFlOps per second (1000 ms) + assertEquals(1000, clock.millis()) } @Test @@ -88,14 +84,10 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) + machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) - // Two sockets with two cores execute 2000 MFlOps per second (500 ms) - assertEquals(500, clock.millis()) - } finally { - machine.close() - } + // Two sockets with two cores execute 2000 MFlOps per second (500 ms) + assertEquals(500, clock.millis()) } @Test @@ -109,16 +101,12 @@ class SimMachineTest { val source = SimPowerSource(engine, capacity = 1000.0) source.connect(machine.psu) - try { - coroutineScope { - launch { machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) } - assertAll( - { assertEquals(100.0, machine.psu.powerDraw) }, - { assertEquals(100.0, source.powerDraw) } - ) - } - } finally { - machine.close() + coroutineScope { + launch { machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) } + assertAll( + { assertEquals(100.0, machine.psu.powerDraw) }, + { assertEquals(100.0, source.powerDraw) } + ) } } @@ -130,22 +118,20 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(object : SimWorkload { - override fun onStart(ctx: SimMachineContext) { - val cpu = ctx.cpus[0] - - cpu.capacity = cpu.model.frequency + 1000.0 - assertEquals(cpu.model.frequency, cpu.capacity) - cpu.capacity = -1.0 - assertEquals(0.0, cpu.capacity) - - ctx.close() - } - }) - } finally { - machine.close() - } + machine.runWorkload(object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + val cpu = ctx.cpus[0] + + cpu.capacity = cpu.model.frequency + 1000.0 + assertEquals(cpu.model.frequency, cpu.capacity) + cpu.capacity = -1.0 + assertEquals(0.0, cpu.capacity) + + ctx.close() + } + + override fun onStop(ctx: SimMachineContext) {} + }) } @Test @@ -156,16 +142,14 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(object : SimWorkload { - override fun onStart(ctx: SimMachineContext) { - assertEquals(32_000 * 4.0, ctx.memory.capacity) - ctx.close() - } - }) - } finally { - machine.close() - } + machine.runWorkload(object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + assertEquals(32_000 * 4.0, ctx.memory.capacity) + ctx.close() + } + + override fun onStop(ctx: SimMachineContext) {} + }) } @Test @@ -176,18 +160,16 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(object : SimWorkload { - override fun onStart(ctx: SimMachineContext) { - val lifecycle = SimWorkloadLifecycle(ctx) - ctx.memory.startConsumer(lifecycle.waitFor(FixedFlowSource(ctx.memory.capacity, utilization = 0.8))) - } - }) - - assertEquals(1250, clock.millis()) - } finally { - machine.close() - } + machine.runWorkload(object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + val lifecycle = SimWorkloadLifecycle(ctx) + ctx.memory.startConsumer(lifecycle.waitFor(FixedFlowSource(ctx.memory.capacity, utilization = 0.8))) + } + + override fun onStop(ctx: SimMachineContext) {} + }) + + assertEquals(1250, clock.millis()) } @Test @@ -202,19 +184,17 @@ class SimMachineTest { val adapter = (machine.peripherals[0] as SimNetworkAdapter) adapter.connect(SimNetworkSink(engine, adapter.bandwidth)) - try { - machine.run(object : SimWorkload { - override fun onStart(ctx: SimMachineContext) { - val lifecycle = SimWorkloadLifecycle(ctx) - val iface = ctx.net[0] - iface.tx.startConsumer(lifecycle.waitFor(FixedFlowSource(iface.bandwidth, utilization = 0.8))) - } - }) - - assertEquals(1250, clock.millis()) - } finally { - machine.close() - } + machine.runWorkload(object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + val lifecycle = SimWorkloadLifecycle(ctx) + val iface = ctx.net[0] + iface.tx.startConsumer(lifecycle.waitFor(FixedFlowSource(iface.bandwidth, utilization = 0.8))) + } + + override fun onStop(ctx: SimMachineContext) {} + }) + + assertEquals(1250, clock.millis()) } @Test @@ -226,19 +206,17 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(object : SimWorkload { - override fun onStart(ctx: SimMachineContext) { - val lifecycle = SimWorkloadLifecycle(ctx) - val disk = ctx.storage[0] - disk.read.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.read.capacity, utilization = 0.8))) - } - }) - - assertEquals(1250, clock.millis()) - } finally { - machine.close() - } + machine.runWorkload(object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + val lifecycle = SimWorkloadLifecycle(ctx) + val disk = ctx.storage[0] + disk.read.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.read.capacity, utilization = 0.8))) + } + + override fun onStop(ctx: SimMachineContext) {} + }) + + assertEquals(1250, clock.millis()) } @Test @@ -250,19 +228,17 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - machine.run(object : SimWorkload { - override fun onStart(ctx: SimMachineContext) { - val lifecycle = SimWorkloadLifecycle(ctx) - val disk = ctx.storage[0] - disk.write.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.write.capacity, utilization = 0.8))) - } - }) - - assertEquals(1250, clock.millis()) - } finally { - machine.close() - } + machine.runWorkload(object : SimWorkload { + override fun onStart(ctx: SimMachineContext) { + val lifecycle = SimWorkloadLifecycle(ctx) + val disk = ctx.storage[0] + disk.write.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.write.capacity, utilization = 0.8))) + } + + override fun onStop(ctx: SimMachineContext) {} + }) + + assertEquals(1250, clock.millis()) } @Test @@ -275,13 +251,11 @@ class SimMachineTest { try { coroutineScope { - launch { machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) } + launch { machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) } cancel() } } catch (_: CancellationException) { // Ignore - } finally { - machine.close() } assertEquals(0, clock.millis()) @@ -295,31 +269,14 @@ class SimMachineTest { SimplePowerDriver(ConstantPowerModel(0.0)) ) - try { - coroutineScope { - launch { - machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) - } + coroutineScope { + launch { + machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) + } - assertThrows<IllegalStateException> { - machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) - } + assertThrows<IllegalStateException> { + machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) } - } finally { - machine.close() } } - - @Test - fun testClose() = runBlockingSimulation { - val machine = SimBareMetalMachine( - FlowEngine(coroutineContext, clock), - machineModel, - SimplePowerDriver(ConstantPowerModel(0.0)) - ) - - machine.close() - assertDoesNotThrow { machine.close() } - assertThrows<IllegalStateException> { machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) } - } } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt index b7f5bf8e..91855e8d 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt @@ -37,6 +37,7 @@ import org.opendc.simulator.compute.model.ProcessingNode import org.opendc.simulator.compute.model.ProcessingUnit import org.opendc.simulator.compute.power.ConstantPowerModel import org.opendc.simulator.compute.power.SimplePowerDriver +import org.opendc.simulator.compute.runWorkload import org.opendc.simulator.compute.workload.SimTrace import org.opendc.simulator.compute.workload.SimTraceFragment import org.opendc.simulator.compute.workload.SimTraceWorkload @@ -80,16 +81,16 @@ internal class SimFairShareHypervisorTest { val hypervisor = SimFairShareHypervisor(platform, null, PerformanceScalingGovernor(), null) launch { - machine.run(hypervisor) + machine.runWorkload(hypervisor) println("Hypervisor finished") } yield() - val vm = hypervisor.createMachine(model) - vm.run(workloadA) + val vm = hypervisor.newMachine(model) + vm.runWorkload(workloadA) yield() - machine.close() + machine.cancel() assertAll( { assertEquals(319781, hypervisor.counters.cpuActiveTime, "Active time does not match") }, @@ -131,22 +132,22 @@ internal class SimFairShareHypervisorTest { val hypervisor = SimFairShareHypervisor(platform, null, null, null) launch { - machine.run(hypervisor) + machine.runWorkload(hypervisor) } yield() coroutineScope { launch { - val vm = hypervisor.createMachine(model) - vm.run(workloadA) - vm.close() + val vm = hypervisor.newMachine(model) + vm.runWorkload(workloadA) + hypervisor.removeMachine(vm) } - val vm = hypervisor.createMachine(model) - vm.run(workloadB) - vm.close() + val vm = hypervisor.newMachine(model) + vm.runWorkload(workloadB) + hypervisor.removeMachine(vm) } yield() - machine.close() + machine.cancel() yield() assertAll( @@ -171,11 +172,11 @@ internal class SimFairShareHypervisorTest { assertDoesNotThrow { launch { - machine.run(hypervisor) + machine.runWorkload(hypervisor) } } - machine.close() + machine.cancel() } @Test @@ -219,20 +220,20 @@ internal class SimFairShareHypervisorTest { ) launch { - machine.run(hypervisor) + machine.runWorkload(hypervisor) } coroutineScope { launch { - val vm = hypervisor.createMachine(model, "a") - vm.run(workloadA) - vm.close() + val vm = hypervisor.newMachine(model, "a") + vm.runWorkload(workloadA) + hypervisor.removeMachine(vm) } - val vm = hypervisor.createMachine(model, "b") - vm.run(workloadB) - vm.close() + val vm = hypervisor.newMachine(model, "b") + vm.runWorkload(workloadB) + hypervisor.removeMachine(vm) } - machine.close() + machine.cancel() } } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt index 02d308ff..823a0ae3 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt @@ -36,6 +36,7 @@ import org.opendc.simulator.compute.model.ProcessingNode import org.opendc.simulator.compute.model.ProcessingUnit import org.opendc.simulator.compute.power.ConstantPowerModel import org.opendc.simulator.compute.power.SimplePowerDriver +import org.opendc.simulator.compute.runWorkload import org.opendc.simulator.compute.workload.* import org.opendc.simulator.core.runBlockingSimulation import org.opendc.simulator.flow.FlowEngine @@ -76,13 +77,13 @@ internal class SimSpaceSharedHypervisorTest { val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) val hypervisor = SimSpaceSharedHypervisor(engine, null, null) - launch { machine.run(hypervisor) } - val vm = hypervisor.createMachine(machineModel) - vm.run(workloadA) + launch { machine.runWorkload(hypervisor) } + val vm = hypervisor.newMachine(machineModel) + vm.runWorkload(workloadA) yield() - vm.close() - machine.close() + hypervisor.removeMachine(vm) + machine.cancel() assertEquals(5 * 60L * 4000, clock.millis()) { "Took enough time" } } @@ -98,12 +99,13 @@ internal class SimSpaceSharedHypervisorTest { val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) val hypervisor = SimSpaceSharedHypervisor(engine, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } yield() - val vm = hypervisor.createMachine(machineModel) - vm.run(workload) - vm.close() - machine.close() + val vm = hypervisor.newMachine(machineModel) + vm.runWorkload(workload) + hypervisor.removeMachine(vm) + + machine.cancel() assertEquals(duration, clock.millis()) { "Took enough time" } } @@ -121,11 +123,11 @@ internal class SimSpaceSharedHypervisorTest { ) val hypervisor = SimSpaceSharedHypervisor(engine, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } yield() - val vm = hypervisor.createMachine(machineModel) - vm.run(workload) - machine.close() + val vm = hypervisor.newMachine(machineModel) + vm.runWorkload(workload) + machine.cancel() assertEquals(duration, clock.millis()) { "Took enough time" } } @@ -142,19 +144,20 @@ internal class SimSpaceSharedHypervisorTest { ) val hypervisor = SimSpaceSharedHypervisor(engine, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } yield() - val vm = hypervisor.createMachine(machineModel) - vm.run(SimRuntimeWorkload(duration)) - vm.close() + val vm = hypervisor.newMachine(machineModel) + vm.runWorkload(SimRuntimeWorkload(duration)) + hypervisor.removeMachine(vm) yield() - val vm2 = hypervisor.createMachine(machineModel) - vm2.run(SimRuntimeWorkload(duration)) - vm2.close() - machine.close() + val vm2 = hypervisor.newMachine(machineModel) + vm2.runWorkload(SimRuntimeWorkload(duration)) + hypervisor.removeMachine(vm2) + + machine.cancel() assertEquals(duration * 2, clock.millis()) { "Took enough time" } } @@ -168,17 +171,17 @@ internal class SimSpaceSharedHypervisorTest { val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) val hypervisor = SimSpaceSharedHypervisor(engine, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } yield() - hypervisor.createMachine(machineModel) + hypervisor.newMachine(machineModel) assertAll( { assertFalse(hypervisor.canFit(machineModel)) }, - { assertThrows<IllegalArgumentException> { hypervisor.createMachine(machineModel) } } + { assertThrows<IllegalArgumentException> { hypervisor.newMachine(machineModel) } } ) - machine.close() + machine.cancel() } /** @@ -192,16 +195,16 @@ internal class SimSpaceSharedHypervisorTest { ) val hypervisor = SimSpaceSharedHypervisor(interpreter, null, null) - launch { machine.run(hypervisor) } + launch { machine.runWorkload(hypervisor) } yield() - hypervisor.createMachine(machineModel).close() + hypervisor.removeMachine(hypervisor.newMachine(machineModel)) assertAll( { assertTrue(hypervisor.canFit(machineModel)) }, - { assertDoesNotThrow { hypervisor.createMachine(machineModel) } } + { assertDoesNotThrow { hypervisor.newMachine(machineModel) } } ) - machine.close() + machine.cancel() } } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt index 574860e8..aa91984a 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt @@ -30,6 +30,7 @@ import org.opendc.simulator.compute.SimBareMetalMachine import org.opendc.simulator.compute.model.* import org.opendc.simulator.compute.power.ConstantPowerModel import org.opendc.simulator.compute.power.SimplePowerDriver +import org.opendc.simulator.compute.runWorkload import org.opendc.simulator.core.runBlockingSimulation import org.opendc.simulator.flow.FlowEngine @@ -67,13 +68,9 @@ class SimTraceWorkloadTest { offset = 0 ) - try { - machine.run(workload) + machine.runWorkload(workload) - assertEquals(4000, clock.millis()) - } finally { - machine.close() - } + assertEquals(4000, clock.millis()) } @Test @@ -94,13 +91,9 @@ class SimTraceWorkloadTest { offset = 1000 ) - try { - machine.run(workload) + machine.runWorkload(workload) - assertEquals(5000, clock.millis()) - } finally { - machine.close() - } + assertEquals(5000, clock.millis()) } @Test @@ -121,14 +114,10 @@ class SimTraceWorkloadTest { offset = 0 ) - try { - delay(1000L) - machine.run(workload) + delay(1000L) + machine.runWorkload(workload) - assertEquals(4000, clock.millis()) - } finally { - machine.close() - } + assertEquals(4000, clock.millis()) } @Test @@ -149,12 +138,8 @@ class SimTraceWorkloadTest { offset = 0 ) - try { - machine.run(workload) + machine.runWorkload(workload) - assertEquals(4000, clock.millis()) - } finally { - machine.close() - } + assertEquals(4000, clock.millis()) } } |