diff options
Diffstat (limited to 'opendc-compute/opendc-compute-workload/src/main')
8 files changed, 109 insertions, 68 deletions
diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt index a802afdb..c9f784ff 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt @@ -31,5 +31,8 @@ public interface ComputeWorkload { /** * Resolve the workload into a list of [VirtualMachine]s to simulate. */ - public fun resolve(loader: ComputeWorkloadLoader, random: RandomGenerator): List<VirtualMachine> + public fun resolve( + loader: ComputeWorkloadLoader, + random: RandomGenerator, + ): List<VirtualMachine> } 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 c5fb3e56..2202f851 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 @@ -29,18 +29,18 @@ import org.opendc.trace.Trace import org.opendc.trace.conv.INTERFERENCE_GROUP_MEMBERS import org.opendc.trace.conv.INTERFERENCE_GROUP_SCORE import org.opendc.trace.conv.INTERFERENCE_GROUP_TARGET -import org.opendc.trace.conv.RESOURCE_CPU_CAPACITY -import org.opendc.trace.conv.RESOURCE_CPU_COUNT -import org.opendc.trace.conv.RESOURCE_ID -import org.opendc.trace.conv.RESOURCE_MEM_CAPACITY -import org.opendc.trace.conv.RESOURCE_START_TIME -import org.opendc.trace.conv.RESOURCE_STATE_CPU_USAGE -import org.opendc.trace.conv.RESOURCE_STATE_DURATION -import org.opendc.trace.conv.RESOURCE_STATE_TIMESTAMP -import org.opendc.trace.conv.RESOURCE_STOP_TIME import org.opendc.trace.conv.TABLE_INTERFERENCE_GROUPS import org.opendc.trace.conv.TABLE_RESOURCES import org.opendc.trace.conv.TABLE_RESOURCE_STATES +import org.opendc.trace.conv.resourceCpuCapacity +import org.opendc.trace.conv.resourceCpuCount +import org.opendc.trace.conv.resourceID +import org.opendc.trace.conv.resourceMemCapacity +import org.opendc.trace.conv.resourceStartTime +import org.opendc.trace.conv.resourceStateCpuUsage +import org.opendc.trace.conv.resourceStateDuration +import org.opendc.trace.conv.resourceStateTimestamp +import org.opendc.trace.conv.resourceStopTime import java.io.File import java.lang.ref.SoftReference import java.time.Duration @@ -71,11 +71,11 @@ public class ComputeWorkloadLoader(private val baseDir: File) { private fun parseFragments(trace: Trace): Map<String, Builder> { val reader = checkNotNull(trace.getTable(TABLE_RESOURCE_STATES)).newReader() - val idCol = reader.resolve(RESOURCE_ID) - val timestampCol = reader.resolve(RESOURCE_STATE_TIMESTAMP) - val durationCol = reader.resolve(RESOURCE_STATE_DURATION) - val coresCol = reader.resolve(RESOURCE_CPU_COUNT) - val usageCol = reader.resolve(RESOURCE_STATE_CPU_USAGE) + val idCol = reader.resolve(resourceID) + val timestampCol = reader.resolve(resourceStateTimestamp) + val durationCol = reader.resolve(resourceStateDuration) + val coresCol = reader.resolve(resourceCpuCount) + val usageCol = reader.resolve(resourceStateCpuUsage) val fragments = mutableMapOf<String, Builder>() @@ -100,15 +100,19 @@ public class ComputeWorkloadLoader(private val baseDir: File) { /** * Read the metadata into a workload. */ - private fun parseMeta(trace: Trace, fragments: Map<String, Builder>, interferenceModel: VmInterferenceModel): List<VirtualMachine> { + private fun parseMeta( + trace: Trace, + fragments: Map<String, Builder>, + interferenceModel: VmInterferenceModel, + ): List<VirtualMachine> { val reader = checkNotNull(trace.getTable(TABLE_RESOURCES)).newReader() - val idCol = reader.resolve(RESOURCE_ID) - val startTimeCol = reader.resolve(RESOURCE_START_TIME) - val stopTimeCol = reader.resolve(RESOURCE_STOP_TIME) - val cpuCountCol = reader.resolve(RESOURCE_CPU_COUNT) - val cpuCapacityCol = reader.resolve(RESOURCE_CPU_CAPACITY) - val memCol = reader.resolve(RESOURCE_MEM_CAPACITY) + val idCol = reader.resolve(resourceID) + val startTimeCol = reader.resolve(resourceStartTime) + val stopTimeCol = reader.resolve(resourceStopTime) + val cpuCountCol = reader.resolve(resourceCpuCount) + val cpuCapacityCol = reader.resolve(resourceCpuCapacity) + val memCol = reader.resolve(resourceMemCapacity) var counter = 0 val entries = mutableListOf<VirtualMachine>() @@ -141,8 +145,8 @@ public class ComputeWorkloadLoader(private val baseDir: File) { submissionTime, endTime, builder.build(), - interferenceModel.getProfile(id) - ) + interferenceModel.getProfile(id), + ), ) } @@ -189,24 +193,28 @@ public class ComputeWorkloadLoader(private val baseDir: File) { /** * Load the trace with the specified [name] and [format]. */ - public fun get(name: String, format: String): List<VirtualMachine> { - val ref = cache.compute(name) { key, oldVal -> - val inst = oldVal?.get() - if (inst == null) { - val path = baseDir.resolve(key) - - logger.info { "Loading trace $key at $path" } - - val trace = Trace.open(path, format) - val fragments = parseFragments(trace) - val interferenceModel = parseInterferenceModel(trace) - val vms = parseMeta(trace, fragments, interferenceModel) - - SoftReference(vms) - } else { - oldVal + public fun get( + name: String, + format: String, + ): List<VirtualMachine> { + val ref = + cache.compute(name) { key, oldVal -> + val inst = oldVal?.get() + if (inst == null) { + val path = baseDir.resolve(key) + + logger.info { "Loading trace $key at $path" } + + val trace = Trace.open(path, format) + val fragments = parseFragments(trace) + val interferenceModel = parseInterferenceModel(trace) + val vms = parseMeta(trace, fragments, interferenceModel) + + SoftReference(vms) + } else { + oldVal + } } - } return checkNotNull(ref?.get()) { "Memory pressure" } } @@ -245,7 +253,12 @@ public class ComputeWorkloadLoader(private val baseDir: File) { * @param usage CPU usage of this fragment. * @param cores Number of cores used. */ - fun add(deadline: Instant, duration: Duration, usage: Double, cores: Int) { + fun add( + deadline: Instant, + duration: Duration, + usage: Double, + cores: Int, + ) { val startTimeMs = (deadline - duration).toEpochMilli() totalLoad += (usage * duration.toMillis()) / 1000.0 // avg MHz * duration = MFLOPs diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloads.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloads.kt index 61a6e3a0..8723f88b 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloads.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloads.kt @@ -32,7 +32,10 @@ import org.opendc.compute.workload.internal.TraceComputeWorkload /** * Construct a workload from a trace. */ -public fun trace(name: String, format: String = "opendc-vm"): ComputeWorkload = TraceComputeWorkload(name, format) +public fun trace( + name: String, + format: String = "opendc-vm", +): ComputeWorkload = TraceComputeWorkload(name, format) /** * Construct a composite workload with the specified fractions. diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt index 622b3c55..deb50f5c 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt @@ -50,5 +50,5 @@ public data class VirtualMachine( val startTime: Instant, val stopTime: Instant, val trace: SimTrace, - val interferenceProfile: VmInterferenceProfile? + val interferenceProfile: VmInterferenceProfile?, ) diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt index 1ac5f4ad..aba493b6 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt @@ -37,7 +37,10 @@ internal class CompositeComputeWorkload(val sources: Map<ComputeWorkload, Double */ private val logger = KotlinLogging.logger {} - override fun resolve(loader: ComputeWorkloadLoader, random: RandomGenerator): List<VirtualMachine> { + override fun resolve( + loader: ComputeWorkloadLoader, + random: RandomGenerator, + ): List<VirtualMachine> { val traces = sources.map { (source, fraction) -> fraction to source.resolve(loader, random) } val totalLoad = traces.sumOf { (_, vms) -> vms.sumOf { it.totalLoad } } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt index fdb599c1..4207b2be 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt @@ -35,7 +35,11 @@ import java.util.random.RandomGenerator * @param fraction The fraction of load/virtual machines to sample * @param sampleLoad A flag to indicate that the sampling should be based on the total load or on the number of VMs. */ -internal class HpcSampledComputeWorkload(val source: ComputeWorkload, val fraction: Double, val sampleLoad: Boolean = false) : ComputeWorkload { +internal class HpcSampledComputeWorkload( + val source: ComputeWorkload, + val fraction: Double, + val sampleLoad: Boolean = false, +) : ComputeWorkload { /** * The logging instance of this class. */ @@ -46,29 +50,35 @@ internal class HpcSampledComputeWorkload(val source: ComputeWorkload, val fracti */ private val pattern = Regex("^(ComputeNode|cn).*") - override fun resolve(loader: ComputeWorkloadLoader, random: RandomGenerator): List<VirtualMachine> { + override fun resolve( + loader: ComputeWorkloadLoader, + random: RandomGenerator, + ): List<VirtualMachine> { val vms = source.resolve(loader, random) - val (hpc, nonHpc) = vms.partition { entry -> - val name = entry.name - name.matches(pattern) - } - - val hpcSequence = generateSequence(0) { it + 1 } - .map { index -> - val res = mutableListOf<VirtualMachine>() - hpc.mapTo(res) { sample(it, index) } - res + val (hpc, nonHpc) = + vms.partition { entry -> + val name = entry.name + name.matches(pattern) } - .flatten() - val nonHpcSequence = generateSequence(0) { it + 1 } - .map { index -> - val res = mutableListOf<VirtualMachine>() - nonHpc.mapTo(res) { sample(it, index) } - res - } - .flatten() + val hpcSequence = + generateSequence(0) { it + 1 } + .map { index -> + val res = mutableListOf<VirtualMachine>() + hpc.mapTo(res) { sample(it, index) } + res + } + .flatten() + + val nonHpcSequence = + generateSequence(0) { it + 1 } + .map { index -> + val res = mutableListOf<VirtualMachine>() + nonHpc.mapTo(res) { sample(it, index) } + res + } + .flatten() logger.debug { "Found ${hpc.size} HPC workloads and ${nonHpc.size} non-HPC workloads" } @@ -135,7 +145,10 @@ internal class HpcSampledComputeWorkload(val source: ComputeWorkload, val fracti /** * Sample a random trace entry. */ - private fun sample(entry: VirtualMachine, i: Int): VirtualMachine { + private fun sample( + entry: VirtualMachine, + i: Int, + ): VirtualMachine { val uid = UUID.nameUUIDFromBytes("${entry.uid}-$i".toByteArray()) return entry.copy(uid = uid) } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt index 6014f37a..c89507fa 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt @@ -37,7 +37,10 @@ internal class LoadSampledComputeWorkload(val source: ComputeWorkload, val fract */ private val logger = KotlinLogging.logger {} - override fun resolve(loader: ComputeWorkloadLoader, random: RandomGenerator): List<VirtualMachine> { + override fun resolve( + loader: ComputeWorkloadLoader, + random: RandomGenerator, + ): List<VirtualMachine> { val vms = source.resolve(loader, random) val res = mutableListOf<VirtualMachine>() diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt index ff88fa3e..39255c59 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt @@ -31,7 +31,10 @@ import java.util.random.RandomGenerator * A [ComputeWorkload] from a trace. */ internal class TraceComputeWorkload(val name: String, val format: String) : ComputeWorkload { - override fun resolve(loader: ComputeWorkloadLoader, random: RandomGenerator): List<VirtualMachine> { + override fun resolve( + loader: ComputeWorkloadLoader, + random: RandomGenerator, + ): List<VirtualMachine> { return loader.get(name, format) } } |