diff options
Diffstat (limited to 'opendc')
10 files changed, 86 insertions, 32 deletions
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/driver/SimpleBareMetalDriver.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/driver/SimpleBareMetalDriver.kt index 4a40dc9f..9396699a 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/driver/SimpleBareMetalDriver.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/driver/SimpleBareMetalDriver.kt @@ -142,7 +142,7 @@ public class SimpleBareMetalDriver( override suspend fun stop(): Node = withContext(domain.coroutineContext) { val node = nodeState.value - if (node.state == NodeState.SHUTOFF) { + if (node.state == NodeState.SHUTOFF || node.state == NodeState.ERROR) { return@withContext node } @@ -298,8 +298,12 @@ public class SimpleBareMetalDriver( get() = domain override suspend fun fail() { - serverContext?.cancel(fail = true) - domain.cancel() + try { + serverContext?.cancel(fail = true) + domain.cancel() + } catch (_: CancellationException) { + // Ignore if the machine has already failed. + } } override fun toString(): String = "SimpleBareMetalDriver(node = ${nodeState.value.uid})" diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/ProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/ProvisioningService.kt index 105505f2..a54d8df4 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/ProvisioningService.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/ProvisioningService.kt @@ -55,6 +55,11 @@ public interface ProvisioningService { public suspend fun deploy(node: Node, image: Image): Node /** + * Stop the specified [Node] . + */ + public suspend fun stop(node: Node): Node + + /** * The service key of this service. */ companion object Key : AbstractServiceKey<ProvisioningService>(UUID.randomUUID(), "provisioner") diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/SimpleProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/SimpleProvisioningService.kt index a7e143aa..f6b236ae 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/SimpleProvisioningService.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/metal/service/SimpleProvisioningService.kt @@ -28,6 +28,7 @@ import com.atlarge.odcsim.Domain import com.atlarge.opendc.compute.core.image.Image import com.atlarge.opendc.compute.metal.Node import com.atlarge.opendc.compute.metal.driver.BareMetalDriver +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.withContext /** @@ -57,4 +58,13 @@ public class SimpleProvisioningService(val domain: Domain) : ProvisioningService val newNode = driver.reboot() return@withContext newNode } + + override suspend fun stop(node: Node): Node = withContext(domain.coroutineContext) { + val driver = nodes[node]!! + try { + driver.stop() + } catch (e: CancellationException) { + node + } + } } diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorEvent.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorEvent.kt index 76a73a57..5c19b00d 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorEvent.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorEvent.kt @@ -64,15 +64,4 @@ public sealed class HypervisorEvent { public val numberOfDeployedImages: Int, public val hostServer: Server ) : HypervisorEvent() - - /** - * This event is emitted when the hypervisor state changes. - * - * @property driver The driver that emitted the event. - * @property server The current server instance. - */ - public data class StateChanged( - override val driver: VirtDriver, - public val server: Server - ) : HypervisorEvent() } diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorImage.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorImage.kt index 657d5067..c21b002d 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorImage.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/HypervisorImage.kt @@ -50,7 +50,6 @@ object HypervisorImage : Image { try { suspendCancellableCoroutine<Unit> {} } finally { - driver.onShutOff() driver.eventFlow.close() } } diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt index b6cf3ac8..3086f4e6 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt @@ -67,7 +67,7 @@ class SimpleVirtDriver( /** * The [Server] on which this hypervisor runs. */ - private val server: Server + val server: Server get() = hostContext.server /** @@ -87,10 +87,6 @@ class SimpleVirtDriver( override val events: Flow<HypervisorEvent> = eventFlow - init { - eventFlow.emit(HypervisorEvent.StateChanged(this, server)) - } - override suspend fun spawn( name: String, image: Image, @@ -243,13 +239,6 @@ class SimpleVirtDriver( } /** - * To be called on shut-off of the hypervisor. - */ - public fun onShutOff() { - eventFlow.emit(HypervisorEvent.StateChanged(this, server)) - } - - /** * A request to schedule a virtual CPU on the host cpu. */ internal data class CpuRequest( diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt index 156521db..b8966275 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt @@ -1,6 +1,7 @@ package com.atlarge.opendc.compute.virt.service import com.atlarge.odcsim.SimulationContext +import com.atlarge.odcsim.flow.EventFlow import com.atlarge.opendc.compute.core.Flavor import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.core.ServerEvent @@ -18,6 +19,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch @@ -52,6 +54,8 @@ class SimpleVirtProvisioningService( */ private val activeImages: MutableSet<ImageView> = mutableSetOf() + override val hypervisorEvents: Flow<HypervisorEvent> = EventFlow() + init { launch { val provisionedNodes = provisioningService.nodes() @@ -84,6 +88,11 @@ class SimpleVirtProvisioningService( } } + override suspend fun terminate() { + val provisionedNodes = provisioningService.nodes() + provisionedNodes.forEach { node -> provisioningService.stop(node) } + } + private var call: Job? = null private fun requestCycle() { diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt index a3ade2fb..fb200f88 100644 --- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt +++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt @@ -3,8 +3,10 @@ package com.atlarge.opendc.compute.virt.service import com.atlarge.opendc.compute.core.Flavor import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.core.image.Image +import com.atlarge.opendc.compute.virt.HypervisorEvent import com.atlarge.opendc.compute.virt.driver.VirtDriver import com.atlarge.opendc.compute.virt.service.allocation.AllocationPolicy +import kotlinx.coroutines.flow.Flow /** * A service for VM provisioning on a cloud. @@ -13,6 +15,11 @@ interface VirtProvisioningService { val allocationPolicy: AllocationPolicy /** + * The events emitted by the hypervisors. + */ + public val hypervisorEvents: Flow<HypervisorEvent> + + /** * Obtain the active hypervisors for this provisioner. */ public suspend fun drivers(): Set<VirtDriver> @@ -25,4 +32,9 @@ interface VirtProvisioningService { * @param flavor The flavor of the machine instance to run this [image] on. */ public suspend fun deploy(name: String, image: Image, flavor: Flavor): Server + + /** + * Terminate the provisioning service releasing all the leased bare-metal machines. + */ + public suspend fun terminate() } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt index 441b4498..120c4f81 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt @@ -22,8 +22,6 @@ class Sc20Monitor( } suspend fun onVmStateChanged(server: Server) { - println("[${simulationContext.clock.millis()}] ${server.uid} ${server.state}") - if (server.state == ServerState.ERROR) { failedInSlice++ } @@ -44,6 +42,8 @@ class Sc20Monitor( ) } + println("[${simulationContext.clock.millis()}] HOST ${server.uid} ${server.state}") + lastServerStates[server] = Pair(server.state, simulationContext.clock.millis()) } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt index d697cde3..ed971df5 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt @@ -28,10 +28,13 @@ import com.atlarge.odcsim.Domain import com.atlarge.odcsim.SimulationEngineProvider import com.atlarge.odcsim.simulationContext import com.atlarge.opendc.compute.core.Flavor +import com.atlarge.opendc.compute.core.Server import com.atlarge.opendc.compute.core.ServerEvent +import com.atlarge.opendc.compute.core.ServerState import com.atlarge.opendc.compute.metal.NODE_CLUSTER import com.atlarge.opendc.compute.metal.service.ProvisioningService import com.atlarge.opendc.compute.virt.HypervisorEvent +import com.atlarge.opendc.compute.virt.driver.SimpleVirtDriver import com.atlarge.opendc.compute.virt.service.SimpleVirtProvisioningService import com.atlarge.opendc.compute.virt.service.allocation.AvailableMemoryAllocationPolicy import com.atlarge.opendc.core.failure.CorrelatedFaultInjector @@ -141,12 +144,28 @@ fun main(args: Array<String>) { // Wait for the hypervisors to be spawned delay(10) + val hypervisors = scheduler.drivers() + var availableHypervisors = hypervisors.size + // Monitor hypervisor events - for (hypervisor in scheduler.drivers()) { + for (hypervisor in hypervisors) { + // TODO Do not expose VirtDriver directly but use Hypervisor class. + monitor.serverStateChanged(hypervisor, (hypervisor as SimpleVirtDriver).server) + hypervisor.server.events + .onEach { event -> + when (event) { + is ServerEvent.StateChanged -> { + monitor.serverStateChanged(hypervisor, event.server) + + if (event.server.state == ServerState.ERROR) + availableHypervisors -= 1 + } + } + } + .launchIn(this) hypervisor.events .onEach { event -> when (event) { - is HypervisorEvent.StateChanged -> monitor.serverStateChanged(event.driver, event.server) is HypervisorEvent.SliceFinished -> monitor.onSliceFinish( simulationContext.clock.millis(), event.requestedBurst, @@ -171,6 +190,9 @@ fun main(args: Array<String>) { } } + val running = mutableSetOf<Server>() + val finish = Channel<Unit>(Channel.RENDEZVOUS) + val reader = Sc20TraceReader(File(traceDirectory), performanceInterferenceModel, getSelectedVmList()) while (reader.hasNext()) { val (time, workload) = reader.next() @@ -181,12 +203,27 @@ fun main(args: Array<String>) { workload.image.name, workload.image, Flavor(workload.image.cores, workload.image.requiredMemory) ) + running += server // Monitor server events - server.events.onEach { if (it is ServerEvent.StateChanged) monitor.onVmStateChanged(it.server) } + server.events + .onEach { + if (it is ServerEvent.StateChanged) + monitor.onVmStateChanged(it.server) + + // Detect whether the VM has finished running + if (it.server.state == ServerState.ERROR || it.server.state == ServerState.SHUTOFF) { + running -= server + + if (running.isEmpty() && (!reader.hasNext() || availableHypervisors == 0)) + finish.send(Unit) + } + } .collect() } } + finish.receive() + scheduler.terminate() println(simulationContext.clock.instant()) } |